diff --git a/consensus/hotstuff/votecollector/statemachine_test.go b/consensus/hotstuff/votecollector/statemachine_test.go index ce0933acd73..4764ac5a535 100644 --- a/consensus/hotstuff/votecollector/statemachine_test.go +++ b/consensus/hotstuff/votecollector/statemachine_test.go @@ -65,7 +65,15 @@ func (s *StateMachineTestSuite) SetupTest() { // prepareMockedProcessor prepares a mocked processor and stores it in map, later it will be used // to mock behavior of verifying vote processor. // Additionally, it setups mocks on the processor assuming the proposer vote will be passed into the processing pipeline. -func (s *StateMachineTestSuite) prepareMockedProcessor(proposal *model.SignedProposal) *mocks.VerifyingVoteProcessor { +// `expectedProposerVoteProcessings` specifies how many `OnVoteProcessed` notifications for the +// proposer's vote the test expects. The proposer's vote is fed into the processing pipeline +// asynchronously, by a worker of the collector's worker pool. Depending on the timing of a +// concurrent transition to the invalid state (e.g. detected proposal equivocation in +// `TestStatus_StateTransitions`), the vote may bypass the verifying processor and be consumed by +// the invalid-state processor instead — in both cases emitting exactly one `OnVoteProcessed` +// notification. Therefore, we register the expectation upfront, rather than lazily in the `Run` +// hook of `Process` (the latter races with the asynchronous vote processing and caused flakiness). +func (s *StateMachineTestSuite) prepareMockedProcessor(proposal *model.SignedProposal, expectedProposerVoteProcessings int) *mocks.VerifyingVoteProcessor { processor := mocks.NewVerifyingVoteProcessor(s.T()) processor.On("Block").Return(func() *model.Block { return proposal.Block @@ -74,9 +82,10 @@ func (s *StateMachineTestSuite) prepareMockedProcessor(proposal *model.SignedPro proposerVote, err := proposal.ProposerVote() require.NoError(s.T(), err) - processor.On("Process", proposerVote).Run(func(_ mock.Arguments) { - s.notifier.On("OnVoteProcessed", proposerVote).Once() - }).Return(nil).Maybe() + processor.On("Process", proposerVote).Return(nil).Maybe() + if expectedProposerVoteProcessings > 0 { + s.notifier.On("OnVoteProcessed", proposerVote).Times(expectedProposerVoteProcessings) + } s.mockedProcessors[proposal.Block.BlockID] = processor return processor @@ -87,7 +96,7 @@ func (s *StateMachineTestSuite) prepareMockedProcessor(proposal *model.SignedPro func (s *StateMachineTestSuite) TestStatus_StateTransitions() { block := helper.MakeBlock(helper.WithBlockView(s.view)) proposal := helper.MakeSignedProposal(helper.WithProposal(helper.MakeProposal(helper.WithBlock(block)))) - s.prepareMockedProcessor(proposal) + s.prepareMockedProcessor(proposal, 1) // by default, we should create in caching status require.Equal(s.T(), hotstuff.VoteCollectorStatusCaching, s.collector.Status()) @@ -124,7 +133,7 @@ func (s *StateMachineTestSuite) Test_FactoryErrorPropagation() { func (s *StateMachineTestSuite) TestAddVote_VerifyingState() { proposal := makeSignedProposalWithView(s.view) block := proposal.Block - processor := s.prepareMockedProcessor(proposal) + processor := s.prepareMockedProcessor(proposal, 1) err := s.collector.ProcessBlock(proposal) require.NoError(s.T(), err) s.T().Run("add-valid-vote", func(t *testing.T) { @@ -213,7 +222,7 @@ func (s *StateMachineTestSuite) TestProcessBlock_ProcessingOfCachedVotes() { votes := 10 proposal := makeSignedProposalWithView(s.view) block := proposal.Block - processor := s.prepareMockedProcessor(proposal) + processor := s.prepareMockedProcessor(proposal, 1) for i := 0; i < votes; i++ { vote := unittest.VoteForBlockFixture(block) // once when caching vote, and once when processing cached vote @@ -250,7 +259,7 @@ func (s *StateMachineTestSuite) TestProcessBlock_ByzantineLeaderEquivocation_Pro block := proposal.Block proposalVote, err := proposal.ProposerVote() require.NoError(s.T(), err) - _ = s.prepareMockedProcessor(proposal) + _ = s.prepareMockedProcessor(proposal, 1) err = s.collector.ProcessBlock(proposal) require.NoError(s.T(), err) @@ -269,10 +278,11 @@ func (s *StateMachineTestSuite) TestProcessBlock_ByzantineLeaderEquivocation_Pro func (s *StateMachineTestSuite) TestProcessBlock_ByzantineLeaderEquivocation_ProposalAfterVote() { proposal := makeSignedProposalWithView(s.view) block := proposal.Block - // in this case proposer vote comes in second and acts as equivocated vote + // in this case proposer vote comes in second and acts as equivocated vote, + // hence it is never processed and no `OnVoteProcessed` notification is expected for it equivocatingVote, err := proposal.ProposerVote() require.NoError(s.T(), err) - processor := s.prepareMockedProcessor(proposal) + processor := s.prepareMockedProcessor(proposal, 0) firstVote := unittest.VoteForBlockFixture(block, unittest.WithVoteSignerID(equivocatingVote.SignerID)) s.notifier.On("OnVoteProcessed", firstVote).Twice() @@ -292,7 +302,7 @@ func (s *StateMachineTestSuite) TestProcessBlock_ByzantineLeaderEquivocation_Pro // Case (2.a): proposal arriving first, stand-alone vote arriving later. func (s *StateMachineTestSuite) TestProcessBlock_ByzantineLeaderSpamming_ProposalBeforeVote() { proposal := makeSignedProposalWithView(s.view) - _ = s.prepareMockedProcessor(proposal) + _ = s.prepareMockedProcessor(proposal, 1) proposalVote, err := proposal.ProposerVote() require.NoError(s.T(), err) @@ -310,11 +320,12 @@ func (s *StateMachineTestSuite) TestProcessBlock_ByzantineLeaderSpamming_Proposa // Case (2.b): stand-alone vote arriving first, proposal arriving second. func (s *StateMachineTestSuite) TestProcessBlock_ByzantineLeaderSpamming_ProposalAfterVote() { proposal := makeSignedProposalWithView(s.view) - _ = s.prepareMockedProcessor(proposal) + // the proposer's vote is processed twice: once when cached as stand-alone vote (arriving first) + // and once when replayed from the cache after the proposal transitioned the collector to verifying + _ = s.prepareMockedProcessor(proposal, 2) proposalVote, err := proposal.ProposerVote() require.NoError(s.T(), err) - s.notifier.On("OnVoteProcessed", proposalVote).Once() err = s.collector.AddVote(proposalVote) require.NoError(s.T(), err) @@ -330,7 +341,7 @@ func (s *StateMachineTestSuite) TestProcessBlock_ByzantineLeaderSpamming_Proposa func (s *StateMachineTestSuite) TestProcessBlock_ByzantineReplicaEquivocation_BeforeProposal() { proposal := makeSignedProposalWithView(s.view) block := proposal.Block - processor := s.prepareMockedProcessor(proposal) + processor := s.prepareMockedProcessor(proposal, 1) vote := unittest.VoteForBlockFixture(block) equivocatingVote := unittest.VoteForBlockFixture(block, unittest.WithVoteSignerID(vote.SignerID)) @@ -356,7 +367,7 @@ func (s *StateMachineTestSuite) TestProcessBlock_ByzantineReplicaEquivocation_Be func (s *StateMachineTestSuite) TestProcessBlock_ByzantineReplicaEquivocation_AfterProposal() { proposal := makeSignedProposalWithView(s.view) block := proposal.Block - processor := s.prepareMockedProcessor(proposal) + processor := s.prepareMockedProcessor(proposal, 1) vote := unittest.VoteForBlockFixture(block) equivocatingVote := unittest.VoteForBlockFixture(block, unittest.WithVoteSignerID(vote.SignerID)) @@ -381,7 +392,7 @@ func (s *StateMachineTestSuite) TestProcessBlock_ByzantineReplicaEquivocation_Af func (s *StateMachineTestSuite) TestProcessBlock_ByzantineReplicaSpamming_BeforeProposal() { proposal := makeSignedProposalWithView(s.view) block := proposal.Block - processor := s.prepareMockedProcessor(proposal) + processor := s.prepareMockedProcessor(proposal, 1) vote := unittest.VoteForBlockFixture(block) @@ -405,7 +416,7 @@ func (s *StateMachineTestSuite) TestProcessBlock_ByzantineReplicaSpamming_Before func (s *StateMachineTestSuite) TestProcessBlock_ByzantineReplicaSpamming_AfterProposal() { proposal := makeSignedProposalWithView(s.view) block := proposal.Block - processor := s.prepareMockedProcessor(proposal) + processor := s.prepareMockedProcessor(proposal, 1) vote := unittest.VoteForBlockFixture(block) err := s.collector.ProcessBlock(proposal) @@ -427,7 +438,7 @@ func (s *StateMachineTestSuite) TestProcessBlock_ByzantineReplicaSpamming_AfterP func (s *StateMachineTestSuite) Test_VoteProcessorErrorPropagation() { proposal := makeSignedProposalWithView(s.view) block := proposal.Block - processor := s.prepareMockedProcessor(proposal) + processor := s.prepareMockedProcessor(proposal, 1) err := s.collector.ProcessBlock(proposal) require.NoError(s.T(), err) @@ -439,18 +450,17 @@ func (s *StateMachineTestSuite) Test_VoteProcessorErrorPropagation() { require.ErrorAs(s.T(), err, &unexpectedError) } -// RegisterVoteConsumer verifies that after registering vote consumer we are receiving all new and past votes +// TestRegisterVoteConsumer verifies that after registering vote consumer we are receiving all new and past votes // in strict ordering of arrival. -func (s *StateMachineTestSuite) RegisterVoteConsumer() { +func (s *StateMachineTestSuite) TestRegisterVoteConsumer() { votes := 10 - proposal := makeSignedProposalWithView(s.view) - block := proposal.Block - processor := s.prepareMockedProcessor(proposal) + block := helper.MakeBlock(helper.WithBlockView(s.view)) expectedVotes := make([]*model.Vote, 0) for i := 0; i < votes; i++ { vote := unittest.VoteForBlockFixture(block) - // eventually it has to be process by processor - processor.On("Process", vote).Return(nil).Once() + // the collector remains in the caching state throughout this test; adding a vote + // there only caches it, emitting an `OnVoteProcessed` notification + s.notifier.On("OnVoteProcessed", vote).Once() require.NoError(s.T(), s.collector.AddVote(vote)) expectedVotes = append(expectedVotes, vote) } @@ -460,12 +470,13 @@ func (s *StateMachineTestSuite) RegisterVoteConsumer() { actualVotes = append(actualVotes, vote) } + // upon registration, the consumer receives all cached votes; subsequently added votes + // are forwarded to the consumer as they arrive s.collector.RegisterVoteConsumer(consumer) for i := 0; i < votes; i++ { vote := unittest.VoteForBlockFixture(block) - // eventually it has to be process by processor - processor.On("Process", vote).Return(nil).Once() + s.notifier.On("OnVoteProcessed", vote).Once() require.NoError(s.T(), s.collector.AddVote(vote)) expectedVotes = append(expectedVotes, vote) } diff --git a/engine/common/follower/compliance_engine_test.go b/engine/common/follower/compliance_engine_test.go index b689d950d7e..7cb46cb3fb7 100644 --- a/engine/common/follower/compliance_engine_test.go +++ b/engine/common/follower/compliance_engine_test.go @@ -200,6 +200,10 @@ func (s *EngineSuite) TestProcessBatchOfDisconnectedBlocks() { // After submitting new finalized block, we check if new batches are filtered based on new finalized view. func (s *EngineSuite) TestProcessFinalizedBlock() { newFinalizedBlock := unittest.BlockHeaderWithParentFixture(s.finalized) + // Ensure a view gap of at least 2 between `s.finalized` and `newFinalizedBlock`, so the child + // block created below can use a view that is lower than `newFinalizedBlock.View` while still + // being greater than its `ParentView` (the fixture picks a random view increment in [1, 10]). + newFinalizedBlock.View = s.finalized.View + 2 done := make(chan struct{}) s.core.On("OnFinalizedBlock", newFinalizedBlock).Run(func(_ mock.Arguments) { @@ -213,7 +217,9 @@ func (s *EngineSuite) TestProcessFinalizedBlock() { // check if batch gets filtered out since it's lower than finalized view done = make(chan struct{}) block := unittest.BlockWithParentFixture(s.finalized) - block.View = newFinalizedBlock.View - 1 // use block view lower than new latest finalized view + // block.View = s.finalized.View + 1, i.e. lower than the new finalized view (s.finalized.View + 2), + // but still greater than block.ParentView (= s.finalized.View), as required for a valid block + block.View = newFinalizedBlock.View - 1 proposal := unittest.ProposalFromBlock(block) diff --git a/engine/common/stop/stop_control_test.go b/engine/common/stop/stop_control_test.go index 75d5c6e8026..0cc6993acba 100644 --- a/engine/common/stop/stop_control_test.go +++ b/engine/common/stop/stop_control_test.go @@ -11,6 +11,7 @@ import ( "github.com/rs/zerolog" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" "github.com/onflow/flow-go/module/irrecoverable" "github.com/onflow/flow-go/utils/unittest" @@ -99,10 +100,18 @@ func TestStopControl_OnProcessedBlock(t *testing.T) { height := uint64(10) // Update processed height and verify it's stored correctly. + // Note: `updateProcessedHeight` hands the height over to a worker routine, which updates + // `lastProcessedHeight` asynchronously. Hence, we wait for the update to take effect. sc.updateProcessedHeight(height) - assert.Equal(t, height, sc.lastProcessedHeight.Value()) + require.Eventually(t, func() bool { + return sc.lastProcessedHeight.Value() == height + }, time.Second, 10*time.Millisecond) // Attempt to set a lower processed height, which should not be allowed. + // Note: `processedHeightChannel` is unbuffered, so once `updateProcessedHeight` returns, the + // worker has received the lower height, implying it has fully processed the previous height. + // Processing the lower height leaves `lastProcessedHeight` unchanged, so reading it here + // concurrently to the worker is safe. sc.updateProcessedHeight(height - 1) assert.Equal(t, height, sc.lastProcessedHeight.Value()) @@ -113,7 +122,9 @@ func TestStopControl_OnProcessedBlock(t *testing.T) { sc.OnVersionUpdate(incompatibleHeight, version) height = incompatibleHeight - 2 sc.updateProcessedHeight(height) - assert.Equal(t, height, sc.lastProcessedHeight.Value()) + require.Eventually(t, func() bool { + return sc.lastProcessedHeight.Value() == height + }, time.Second, 10*time.Millisecond) // Prepare to trigger the Throw method when the incompatible block height is processed. height = incompatibleHeight - 1 diff --git a/engine/common/synchronization/engine_spam_test.go b/engine/common/synchronization/engine_spam_test.go index 91037a3c3ed..273496aeef4 100644 --- a/engine/common/synchronization/engine_spam_test.go +++ b/engine/common/synchronization/engine_spam_test.go @@ -26,13 +26,43 @@ func (ss *SyncSuite) TestLoad_Process_SyncRequest_HigherThanReceiver_OutsideTole ctx, cancel := irrecoverable.NewMockSignalerContextWithCancel(ss.T(), context.Background()) ss.e.Start(ctx) unittest.AssertClosesBefore(ss.T(), ss.e.Ready(), time.Second) - defer cancel() + // Stop the engine and wait for its worker routines to exit before the test returns. Otherwise, + // leaked workers would access suite fields, racing with the next test's SetupTest. + defer func() { + cancel() + unittest.AssertClosesBefore(ss.T(), ss.e.Done(), time.Second) + }() load := 1000 // reset misbehavior report counter for each subtest misbehaviorsCounter := 0 + // if request height is higher than local finalized, we should not respond + reqHeight := ss.head.Height + 1 + + // Register loop-invariant mock expectations once, before the load loop. Registering them per + // iteration accumulates thousands of expected-call entries, which makes the mock's call matching + // quadratic in the load and slows this test down drastically. + ss.core.On("HandleHeight", ss.head, reqHeight) + ss.core.On("WithinTolerance", ss.head, reqHeight).Return(false) + + // maybe function calls that might or might not occur over the course of the load test + ss.core.On("ScanPending", ss.head).Return([]chainsync.Range{}, []chainsync.Batch{}).Maybe() + ss.con.On("Multicast", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe() + + // count misbehavior reports over the course of a load test + ss.con.On("ReportMisbehavior", mock.Anything).Return(mock.Anything).Run( + func(args mock.Arguments) { + misbehaviorsCounter++ + }, + ) + + // force creating misbehavior report by setting syncRequestProb to 1.0 (i.e. report misbehavior 100% of the time) + ss.e.spamDetectionConfig.syncRequestProb = 1.0 + + ss.metrics.On("MessageReceived", metrics.EngineSynchronization, metrics.MessageSyncRequest).Times(load) + for i := 0; i < load; i++ { // generate origin and request message originID := unittest.IdentifierFixture() @@ -42,34 +72,13 @@ func (ss *SyncSuite) TestLoad_Process_SyncRequest_HigherThanReceiver_OutsideTole req := &flow.SyncRequest{ Nonce: nonce, - Height: 0, + Height: reqHeight, } - // if request height is higher than local finalized, we should not respond - req.Height = ss.head.Height + 1 - - ss.core.On("HandleHeight", ss.head, req.Height) - ss.core.On("WithinTolerance", ss.head, req.Height).Return(false) - ss.con.AssertNotCalled(ss.T(), "Unicast", mock.Anything, mock.Anything) - - // maybe function calls that might or might not occur over the course of the load test - ss.core.On("ScanPending", ss.head).Return([]chainsync.Range{}, []chainsync.Batch{}).Maybe() - ss.con.On("Multicast", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe() - - // count misbehavior reports over the course of a load test - ss.con.On("ReportMisbehavior", mock.Anything).Return(mock.Anything).Run( - func(args mock.Arguments) { - misbehaviorsCounter++ - }, - ) - - // force creating misbehavior report by setting syncRequestProb to 1.0 (i.e. report misbehavior 100% of the time) - ss.e.spamDetectionConfig.syncRequestProb = 1.0 - - ss.metrics.On("MessageReceived", metrics.EngineSynchronization, metrics.MessageSyncRequest).Once() require.NoError(ss.T(), ss.e.Process(channels.SyncCommittee, originID, req)) } + ss.con.AssertNotCalled(ss.T(), "Unicast", mock.Anything, mock.Anything) ss.core.AssertExpectations(ss.T()) ss.con.AssertExpectations(ss.T()) ss.metrics.AssertExpectations(ss.T()) @@ -84,7 +93,12 @@ func (ss *SyncSuite) TestLoad_Process_SyncRequest_HigherThanReceiver_OutsideTole ctx, cancel := irrecoverable.NewMockSignalerContextWithCancel(ss.T(), context.Background()) ss.e.Start(ctx) unittest.AssertClosesBefore(ss.T(), ss.e.Ready(), time.Second) - defer cancel() + // Stop the engine and wait for its worker routines to exit before the test returns. Otherwise, + // leaked workers would access suite fields, racing with the next test's SetupTest. + defer func() { + cancel() + unittest.AssertClosesBefore(ss.T(), ss.e.Done(), time.Second) + }() load := 1000 @@ -103,28 +117,55 @@ func (ss *SyncSuite) TestLoad_Process_SyncRequest_HigherThanReceiver_OutsideTole // expect to never get misbehavior report loadGroups = append(loadGroups, loadGroup{0.0, 0, 0}) + // The lower and upper bounds below are the exact quantiles of the Binomial(1000, p) distribution + // of the misbehavior report count, such that the probability of the count falling outside the + // bounds is at most 1e-9 per tail. This keeps the test meaningful while making failures of a + // correct implementation practically impossible (previous, tighter bounds caused flakiness). + // expect to get misbehavior report about 0.1% of the time (1 in 1000 requests) - loadGroups = append(loadGroups, loadGroup{0.001, 0, 7}) + loadGroups = append(loadGroups, loadGroup{0.001, 0, 11}) // expect to get misbehavior report about 1% of the time - loadGroups = append(loadGroups, loadGroup{0.01, 5, 15}) + loadGroups = append(loadGroups, loadGroup{0.01, 0, 34}) // expect to get misbehavior report about 10% of the time - loadGroups = append(loadGroups, loadGroup{0.1, 75, 140}) + loadGroups = append(loadGroups, loadGroup{0.1, 48, 161}) // expect to get misbehavior report about 50% of the time - loadGroups = append(loadGroups, loadGroup{0.5, 450, 550}) + loadGroups = append(loadGroups, loadGroup{0.5, 405, 595}) // expect to get misbehavior report about 90% of the time - loadGroups = append(loadGroups, loadGroup{0.9, 850, 950}) + loadGroups = append(loadGroups, loadGroup{0.9, 839, 952}) // reset misbehavior report counter for each subtest misbehaviorsCounter := 0 for _, loadGroup := range loadGroups { ss.T().Run(fmt.Sprintf("load test; pfactor=%f lower=%d upper=%d", loadGroup.syncRequestProbabilityFactor, loadGroup.expectedMisbehaviorsLower, loadGroup.expectedMisbehaviorsUpper), func(t *testing.T) { + // if request height is higher than local finalized, we should not respond + reqHeight := ss.head.Height + 1 + + // Register loop-invariant mock expectations once, before the load loop. Registering them per + // iteration accumulates thousands of expected-call entries, which makes the mock's call matching + // quadratic in the load and slows this test down drastically. + ss.core.On("HandleHeight", ss.head, reqHeight) + ss.core.On("WithinTolerance", ss.head, reqHeight).Return(false) + + // maybe function calls that might or might not occur over the course of the load test + ss.core.On("ScanPending", ss.head).Return([]chainsync.Range{}, []chainsync.Batch{}).Maybe() + ss.con.On("Multicast", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe() + + // count misbehavior reports over the course of a load test + ss.con.On("ReportMisbehavior", mock.Anything).Return(mock.Anything).Maybe().Run( + func(args mock.Arguments) { + misbehaviorsCounter++ + }, + ) + ss.e.spamDetectionConfig.syncRequestProb = loadGroup.syncRequestProbabilityFactor + ss.metrics.On("MessageSent", metrics.EngineSynchronization, metrics.MessageSyncRequest).Maybe() + ss.metrics.On("MessageReceived", metrics.EngineSynchronization, metrics.MessageSyncRequest).Times(load) + for i := 0; i < load; i++ { - ss.T().Log("load iteration", i) nonce, err := rand.Uint64() require.NoError(ss.T(), err, "should generate nonce") @@ -132,32 +173,14 @@ func (ss *SyncSuite) TestLoad_Process_SyncRequest_HigherThanReceiver_OutsideTole originID := unittest.IdentifierFixture() req := &flow.SyncRequest{ Nonce: nonce, - Height: 0, + Height: reqHeight, } - // if request height is higher than local finalized, we should not respond - req.Height = ss.head.Height + 1 - - ss.core.On("HandleHeight", ss.head, req.Height) - ss.core.On("WithinTolerance", ss.head, req.Height).Return(false) - ss.con.AssertNotCalled(ss.T(), "Unicast", mock.Anything, mock.Anything) - - // maybe function calls that might or might not occur over the course of the load test - ss.core.On("ScanPending", ss.head).Return([]chainsync.Range{}, []chainsync.Batch{}).Maybe() - ss.con.On("Multicast", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe() - - // count misbehavior reports over the course of a load test - ss.con.On("ReportMisbehavior", mock.Anything).Return(mock.Anything).Maybe().Run( - func(args mock.Arguments) { - misbehaviorsCounter++ - }, - ) - ss.e.spamDetectionConfig.syncRequestProb = loadGroup.syncRequestProbabilityFactor - ss.metrics.On("MessageSent", metrics.EngineSynchronization, metrics.MessageSyncRequest).Maybe() - ss.metrics.On("MessageReceived", metrics.EngineSynchronization, metrics.MessageSyncRequest).Once() require.NoError(ss.T(), ss.e.Process(channels.SyncCommittee, originID, req)) } + ss.con.AssertNotCalled(ss.T(), "Unicast", mock.Anything, mock.Anything) + // check function call expectations at the end of the load test; otherwise, load test would take much longer ss.core.AssertExpectations(ss.T()) ss.con.AssertExpectations(ss.T()) @@ -181,7 +204,12 @@ func (ss *SyncSuite) TestLoad_Process_RangeRequest_SometimesReportSpam() { ctx, cancel := irrecoverable.NewMockSignalerContextWithCancel(ss.T(), context.Background()) ss.e.Start(ctx) unittest.AssertClosesBefore(ss.T(), ss.e.Ready(), time.Second) - defer cancel() + // Stop the engine and wait for its worker routines to exit before the test returns. Otherwise, + // leaked workers would access suite fields, racing with the next test's SetupTest. + defer func() { + cancel() + unittest.AssertClosesBefore(ss.T(), ss.e.Done(), time.Second) + }() load := 1000 @@ -199,32 +227,35 @@ func (ss *SyncSuite) TestLoad_Process_RangeRequest_SometimesReportSpam() { loadGroups := []loadGroup{} - // using a very small range (1) with a 10% base probability factor, expect to almost never get misbehavior report, about 0.003% of the time (3 in 1000 requests) + // The lower and upper bounds below are the exact quantiles of the Binomial(1000, p) distribution + // of the misbehavior report count (with p being the expected probability factor), such that the + // probability of the count falling outside the bounds is at most 1e-9 per tail. This keeps the + // test meaningful while making failures of a correct implementation practically impossible + // (previous, tighter bounds caused flakiness). + + // using a very small range (1) with a 10% base probability factor, expect to almost never get misbehavior report, about 0.3% of the time (3 in 1000 requests) // expected probability factor: 0.1 * ((10-9) + 1)/64 = 0.003125 - loadGroups = append(loadGroups, loadGroup{0.1, 0, 15, 9, 10}) + loadGroups = append(loadGroups, loadGroup{0.1, 0, 18, 9, 10}) // using a small range (10) with a 10% base probability factor, expect to get misbehavior report about 1.7% of the time (17 in 1000 requests) // expected probability factor: 0.1 * ((11-1) + 1)/64 = 0.0171875 - loadGroups = append(loadGroups, loadGroup{0.1, 5, 31, 1, 11}) + loadGroups = append(loadGroups, loadGroup{0.1, 0, 47, 1, 11}) // using a large range (99) with a 10% base probability factor, expect to get misbehavior report about 15% of the time (150 in 1000 requests) // expected probability factor: 0.1 * ((100-1) + 1)/64 = 0.15625 - loadGroups = append(loadGroups, loadGroup{0.1, 110, 200, 1, 100}) + loadGroups = append(loadGroups, loadGroup{0.1, 92, 229, 1, 100}) // using a flat range (0) (from height == to height) with a 1% base probability factor, expect to almost never get a misbehavior report, about 0.16% of the time (2 in 1000 requests) // expected probability factor: 0.01 * ((1-1) + 1)/64 = 0.0015625 - // Note: the expected upper misbehavior count is 5 even though the expected probability is close to 0 to cover outlier cases during the load test to avoid flakiness in CI. - // Due of the probabilistic nature of the load tests, you sometimes get edge cases (that cover outliers where out of a 1000 messages, up to 5 could be reported as spam. - // 5/1000 = 0.005 and the calculated probability is 0.00171875 which 2.9x as small. - loadGroups = append(loadGroups, loadGroup{0.01, 0, 5, 1, 1}) + loadGroups = append(loadGroups, loadGroup{0.01, 0, 14, 1, 1}) // using a small range (10) with a 1% base probability factor, expect to almost never get misbehavior report, about 0.17% of the time (2 in 1000 requests) // expected probability factor: 0.01 * ((11-1) + 1)/64 = 0.00171875 - loadGroups = append(loadGroups, loadGroup{0.01, 0, 7, 1, 11}) + loadGroups = append(loadGroups, loadGroup{0.01, 0, 14, 1, 11}) // using a very large range (999) with a 1% base probability factor, expect to get misbehavior report about 15% of the time (150 in 1000 requests) // expected probability factor: 0.01 * ((1000-1) + 1)/64 = 0.15625 - loadGroups = append(loadGroups, loadGroup{0.01, 110, 200, 1, 1000}) + loadGroups = append(loadGroups, loadGroup{0.01, 92, 229, 1, 1000}) // ALWAYS REPORT SPAM FOR INVALID RANGE REQUESTS OR RANGE REQUESTS THAT ARE FAR OUTSIDE OF THE TOLERANCE @@ -238,10 +269,27 @@ func (ss *SyncSuite) TestLoad_Process_RangeRequest_SometimesReportSpam() { // reset misbehavior report counter for each subtest misbehaviorsCounter := 0 + // Register loop-invariant mock expectations once, before the load loops. Registering them per + // iteration accumulates thousands of expected-call entries, which makes the mock's call matching + // quadratic in the load and slows this test down drastically. + + // maybe function calls that might or might not occur over the course of the load test + ss.core.On("ScanPending", ss.head).Return([]chainsync.Range{}, []chainsync.Batch{}).Maybe() + ss.con.On("Multicast", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe() + ss.metrics.On("MessageSent", metrics.EngineSynchronization, metrics.MessageSyncRequest).Maybe() + + // count misbehavior reports over the course of a load test + ss.con.On("ReportMisbehavior", mock.Anything).Return(mock.Anything).Maybe().Run( + func(args mock.Arguments) { + misbehaviorsCounter++ + }, + ) + for _, loadGroup := range loadGroups { - for i := 0; i < load; i++ { - ss.T().Log("load iteration", i) + ss.e.spamDetectionConfig.rangeRequestBaseProb = loadGroup.rangeRequestBaseProb + ss.metrics.On("MessageReceived", metrics.EngineSynchronization, metrics.MessageRangeRequest).Times(load) + for i := 0; i < load; i++ { nonce, err := rand.Uint64() require.NoError(ss.T(), err, "should generate nonce") @@ -253,19 +301,6 @@ func (ss *SyncSuite) TestLoad_Process_RangeRequest_SometimesReportSpam() { ToHeight: loadGroup.toHeight, } - // maybe function calls that might or might not occur over the course of the load test - ss.core.On("ScanPending", ss.head).Return([]chainsync.Range{}, []chainsync.Batch{}).Maybe() - ss.con.On("Multicast", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe() - - // count misbehavior reports over the course of a load test - ss.con.On("ReportMisbehavior", mock.Anything).Return(mock.Anything).Maybe().Run( - func(args mock.Arguments) { - misbehaviorsCounter++ - }, - ) - ss.e.spamDetectionConfig.rangeRequestBaseProb = loadGroup.rangeRequestBaseProb - ss.metrics.On("MessageReceived", metrics.EngineSynchronization, metrics.MessageRangeRequest).Once() - ss.metrics.On("MessageSent", metrics.EngineSynchronization, metrics.MessageSyncRequest).Maybe() require.NoError(ss.T(), ss.e.Process(channels.SyncCommittee, originID, req)) } // check function call expectations at the end of the load test; otherwise, load test would take much longer @@ -290,7 +325,12 @@ func (ss *SyncSuite) TestLoad_Process_BatchRequest_SometimesReportSpam() { ctx, cancel := irrecoverable.NewMockSignalerContextWithCancel(ss.T(), context.Background()) ss.e.Start(ctx) unittest.AssertClosesBefore(ss.T(), ss.e.Ready(), time.Second) - defer cancel() + // Stop the engine and wait for its worker routines to exit before the test returns. Otherwise, + // leaked workers would access suite fields, racing with the next test's SetupTest. + defer func() { + cancel() + unittest.AssertClosesBefore(ss.T(), ss.e.Done(), time.Second) + }() load := 1000 @@ -307,25 +347,31 @@ func (ss *SyncSuite) TestLoad_Process_BatchRequest_SometimesReportSpam() { loadGroups := []loadGroup{} - // using a very small batch request (1 block ID) with a 10% base probability factor, expect to almost never get misbehavior report, about 0.003% of the time (3 in 1000 requests) - // expected probability factor: 0.1 * ((10-9) + 1)/64 = 0.003125 - loadGroups = append(loadGroups, loadGroup{0.1, 0, 15, repeatedBlockIDs(1)}) + // The lower and upper bounds below are the exact quantiles of the Binomial(1000, p) distribution + // of the misbehavior report count (with p being the expected probability factor), such that the + // probability of the count falling outside the bounds is at most 1e-9 per tail. This keeps the + // test meaningful while making failures of a correct implementation practically impossible + // (previous, tighter bounds caused flakiness). + + // using a very small batch request (1 block ID) with a 10% base probability factor, expect to almost never get misbehavior report, about 0.3% of the time (3 in 1000 requests) + // expected probability factor: 0.1 * (1 + 1)/64 = 0.003125 + loadGroups = append(loadGroups, loadGroup{0.1, 0, 18, repeatedBlockIDs(1)}) // using a small batch request (10 block IDs) with a 10% base probability factor, expect to get misbehavior report about 1.7% of the time (17 in 1000 requests) - // expected probability factor: 0.1 * ((11-1) + 1)/64 = 0.0171875 - loadGroups = append(loadGroups, loadGroup{0.1, 5, 31, repeatedBlockIDs(10)}) + // expected probability factor: 0.1 * (10 + 1)/64 = 0.0171875 + loadGroups = append(loadGroups, loadGroup{0.1, 0, 47, repeatedBlockIDs(10)}) // using a large batch request (99 block IDs) with a 10% base probability factor, expect to get misbehavior report about 15% of the time (150 in 1000 requests) - // expected probability factor: 0.1 * ((100-1) + 1)/64 = 0.15625 - loadGroups = append(loadGroups, loadGroup{0.1, 110, 200, repeatedBlockIDs(99)}) + // expected probability factor: 0.1 * (99 + 1)/64 = 0.15625 + loadGroups = append(loadGroups, loadGroup{0.1, 92, 229, repeatedBlockIDs(99)}) // using a small batch request (10 block IDs) with a 1% base probability factor, expect to almost never get misbehavior report, about 0.17% of the time (2 in 1000 requests) - // expected probability factor: 0.01 * ((11-1) + 1)/64 = 0.00171875 - loadGroups = append(loadGroups, loadGroup{0.01, 0, 7, repeatedBlockIDs(10)}) + // expected probability factor: 0.01 * (10 + 1)/64 = 0.00171875 + loadGroups = append(loadGroups, loadGroup{0.01, 0, 14, repeatedBlockIDs(10)}) // using a very large batch request (999 block IDs) with a 1% base probability factor, expect to get misbehavior report about 15% of the time (150 in 1000 requests) - // expected probability factor: 0.01 * ((1000-1) + 1)/64 = 0.15625 - loadGroups = append(loadGroups, loadGroup{0.01, 110, 200, repeatedBlockIDs(999)}) + // expected probability factor: 0.01 * (999 + 1)/64 = 0.15625 + loadGroups = append(loadGroups, loadGroup{0.01, 92, 229, repeatedBlockIDs(999)}) // ALWAYS REPORT SPAM FOR INVALID BATCH REQUESTS OR BATCH REQUESTS THAT ARE FAR OUTSIDE OF THE TOLERANCE @@ -338,10 +384,28 @@ func (ss *SyncSuite) TestLoad_Process_BatchRequest_SometimesReportSpam() { // reset misbehavior report counter for each subtest misbehaviorsCounter := 0 + + // Register loop-invariant mock expectations once, before the load loops. Registering them per + // iteration accumulates thousands of expected-call entries, which makes the mock's call matching + // quadratic in the load and slows this test down drastically. + + // maybe function calls that might or might not occur over the course of the load test + ss.core.On("ScanPending", ss.head).Return([]chainsync.Range{}, []chainsync.Batch{}).Maybe() + ss.con.On("Multicast", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe() + ss.metrics.On("MessageSent", metrics.EngineSynchronization, metrics.MessageSyncRequest).Maybe() + + // count misbehavior reports over the course of a load test + ss.con.On("ReportMisbehavior", mock.Anything).Return(mock.Anything).Maybe().Run( + func(args mock.Arguments) { + misbehaviorsCounter++ + }, + ) + for _, loadGroup := range loadGroups { - for i := 0; i < load; i++ { - ss.T().Log("load iteration", i) + ss.e.spamDetectionConfig.batchRequestBaseProb = loadGroup.batchRequestBaseProb + ss.metrics.On("MessageReceived", metrics.EngineSynchronization, metrics.MessageBatchRequest).Times(load) + for i := 0; i < load; i++ { nonce, err := rand.Uint64() require.NoError(ss.T(), err, "should generate nonce") @@ -352,20 +416,6 @@ func (ss *SyncSuite) TestLoad_Process_BatchRequest_SometimesReportSpam() { BlockIDs: loadGroup.blockIDs, } - // maybe function calls that might or might not occur over the course of the load test - ss.core.On("ScanPending", ss.head).Return([]chainsync.Range{}, []chainsync.Batch{}).Maybe() - ss.con.On("Multicast", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe() - - // count misbehavior reports over the course of a load test - ss.con.On("ReportMisbehavior", mock.Anything).Return(mock.Anything).Maybe().Run( - func(args mock.Arguments) { - misbehaviorsCounter++ - }, - ) - ss.e.spamDetectionConfig.batchRequestBaseProb = loadGroup.batchRequestBaseProb - ss.metrics.On("MessageSent", metrics.EngineSynchronization, metrics.MessageSyncRequest).Maybe() - ss.metrics.On("MessageReceived", metrics.EngineSynchronization, metrics.MessageBatchRequest).Once() - require.NoError(ss.T(), ss.e.Process(channels.SyncCommittee, originID, req)) } // check function call expectations at the end of the load test; otherwise, load test would take much longer diff --git a/fvm/evm/evm_test.go b/fvm/evm/evm_test.go index 2177c2b54d0..16525993fc6 100644 --- a/fvm/evm/evm_test.go +++ b/fvm/evm/evm_test.go @@ -2943,7 +2943,12 @@ func TestCadenceOwnedAccountFunctionalities(t *testing.T) { require.NoError(t, err) require.NoError(t, output.Err) assert.Len(t, output.Events, 3) - assert.Len(t, state.UpdatedRegisterIDs(), 13) + // This transaction must update state. The exact number of updated registers varies + // slightly between runs: the test environment uses a random block ID, which selects + // the UUID partition register (`uuid` vs `uuid_N`) and thereby also the atree slab + // layout. Hence, we only check that state was updated at all - in contrast to + // `dryCall` below, which must not update any registers. + assert.NotEmpty(t, state.UpdatedRegisterIDs()) assert.Equal( t, flow.EventType("A.f8d6e0586b0a20c7.EVM.TransactionExecuted"), @@ -3091,7 +3096,12 @@ func TestCadenceOwnedAccountFunctionalities(t *testing.T) { require.NoError(t, err) require.NoError(t, output.Err) assert.Len(t, output.Events, 3) - assert.Len(t, state.UpdatedRegisterIDs(), 13) + // This transaction must update state. The exact number of updated registers varies + // slightly between runs: the test environment uses a random block ID, which selects + // the UUID partition register (`uuid` vs `uuid_N`) and thereby also the atree slab + // layout. Hence, we only check that state was updated at all - in contrast to + // `dryCall` below, which must not update any registers. + assert.NotEmpty(t, state.UpdatedRegisterIDs()) assert.Equal( t, flow.EventType("A.f8d6e0586b0a20c7.EVM.TransactionExecuted"), diff --git a/module/builder/collection/builder_test.go b/module/builder/collection/builder_test.go index a63e3583e81..854190decec 100644 --- a/module/builder/collection/builder_test.go +++ b/module/builder/collection/builder_test.go @@ -414,16 +414,21 @@ func (suite *BuilderSuite) TestBuildOn_WithOrphanedReferenceBlock() { protocolStateID := protocolState.ID() // create a block extending genesis which will be orphaned - orphan := unittest.BlockWithParentAndPayload( + // Since `orphan` and `block1` are siblings, they must have different views. We track the views + // used so far in `usedViews` to prevent the fixtures from generating two blocks with the same view. + usedViews := make(map[uint64]struct{}) + orphan := unittest.BlockWithParentAndPayloadAndUniqueView( genesis, unittest.PayloadFixture(unittest.WithProtocolStateID(protocolStateID)), + usedViews, ) err = suite.protoState.ExtendCertified(context.Background(), unittest.NewCertifiedBlock(orphan)) suite.Require().NoError(err) // create and finalize a block on top of genesis, orphaning `orphan` - block1 := unittest.BlockWithParentAndPayload( + block1 := unittest.BlockWithParentAndPayloadAndUniqueView( genesis, unittest.PayloadFixture(unittest.WithProtocolStateID(protocolStateID)), + usedViews, ) err = suite.protoState.ExtendCertified(context.Background(), unittest.NewCertifiedBlock(block1)) suite.Require().NoError(err) diff --git a/module/executiondatasync/optimistic_sync/pipeline/pipeline_functional_test.go b/module/executiondatasync/optimistic_sync/pipeline/pipeline_functional_test.go index c7da63bec2f..bf73404858f 100644 --- a/module/executiondatasync/optimistic_sync/pipeline/pipeline_functional_test.go +++ b/module/executiondatasync/optimistic_sync/pipeline/pipeline_functional_test.go @@ -249,11 +249,11 @@ func (p *PipelineFunctionalSuite) TestPipelineCompletesSuccessfully() { pipeline.OnParentStateUpdated(optimistic_sync.StateComplete) - waitForStateUpdates(p.T(), updateChan, errChan, optimistic_sync.StateProcessing, optimistic_sync.StateWaitingPersist) + waitForStateUpdatesAndNoError(p.T(), updateChan, errChan, optimistic_sync.StateProcessing, optimistic_sync.StateWaitingPersist) pipeline.SetSealed() - waitForStateUpdates(p.T(), updateChan, errChan, optimistic_sync.StateComplete) + waitForStateUpdatesAndNoError(p.T(), updateChan, errChan, optimistic_sync.StateComplete) actualEvents, err := p.persistentEvents.ByBlockID(p.block.ID()) p.Require().NoError(err) @@ -370,7 +370,8 @@ func (p *PipelineFunctionalSuite) TestPipelinePersistingError() { p.WithRunningPipeline(func(pipeline optimistic_sync.Pipeline, updateChan chan optimistic_sync.State, errChan chan error, cancel context.CancelFunc) { pipeline.OnParentStateUpdated(optimistic_sync.StateComplete) - waitForStateUpdates(p.T(), updateChan, errChan, optimistic_sync.StateProcessing, optimistic_sync.StateWaitingPersist) + // no error may occur before SetSealed triggers the persist step + waitForStateUpdatesAndNoError(p.T(), updateChan, errChan, optimistic_sync.StateProcessing, optimistic_sync.StateWaitingPersist) pipeline.SetSealed() @@ -440,7 +441,8 @@ func (p *PipelineFunctionalSuite) TestMainCtxCancellationDuringWaitingPersist() p.WithRunningPipeline(func(pipeline optimistic_sync.Pipeline, updateChan chan optimistic_sync.State, errChan chan error, cancel context.CancelFunc) { pipeline.OnParentStateUpdated(optimistic_sync.StateComplete) - waitForStateUpdates(p.T(), updateChan, errChan, optimistic_sync.StateProcessing, optimistic_sync.StateWaitingPersist) + // no error may occur before the context is cancelled below + waitForStateUpdatesAndNoError(p.T(), updateChan, errChan, optimistic_sync.StateProcessing, optimistic_sync.StateWaitingPersist) cancel() diff --git a/module/executiondatasync/optimistic_sync/pipeline/pipeline_test_utils.go b/module/executiondatasync/optimistic_sync/pipeline/pipeline_test_utils.go index 95f68283742..0d028de0802 100644 --- a/module/executiondatasync/optimistic_sync/pipeline/pipeline_test_utils.go +++ b/module/executiondatasync/optimistic_sync/pipeline/pipeline_test_utils.go @@ -1,6 +1,7 @@ package pipeline import ( + "sync/atomic" "testing" "testing/synctest" "time" @@ -16,29 +17,30 @@ import ( // mockStateProvider is a mock implementation of a parent state provider. // It tracks the current state and notifies the pipeline when the state changes. +// The state is stored atomically, since it is updated by the test goroutine while +// being read concurrently by the pipeline's worker. type mockStateProvider struct { - state optimistic_sync.State + state atomic.Int32 } var _ optimistic_sync.PipelineStateProvider = (*mockStateProvider)(nil) // NewMockStateProvider initializes a mockStateProvider with the default state StatePending. func NewMockStateProvider() *mockStateProvider { - return &mockStateProvider{ - state: optimistic_sync.StatePending, - } + m := &mockStateProvider{} + m.state.Store(int32(optimistic_sync.StatePending)) + return m } // UpdateState sets the internal state and triggers a pipeline update. func (m *mockStateProvider) UpdateState(state optimistic_sync.State, pipeline *Pipeline) { - m.state = state + m.state.Store(int32(state)) pipeline.OnParentStateUpdated(state) } // GetState returns the current internal state. - func (m *mockStateProvider) GetState() optimistic_sync.State { - return m.state + return optimistic_sync.State(m.state.Load()) } // mockStateConsumer is a mock implementation used in tests to receive state updates from the pipeline. @@ -66,6 +68,18 @@ func waitForStateUpdates(t *testing.T, updateChan <-chan optimistic_sync.State, done := make(chan struct{}) unittest.RequireReturnsBefore(t, func() { for _, expected := range expectedStates { + // Prefer consuming pending state updates over errors: the pipeline always reports a state + // update before emitting an error, but both may already be queued by the time we read + // them (e.g. in tests expecting a state transition immediately followed by an expected + // error). A single select would pick one of the ready channels at random, potentially + // failing on the error before consuming the preceding state update. + select { + case update := <-updateChan: + assert.Equalf(t, expected, update, "expected pipeline to transition to %s, but got %s", expected, update) + continue + default: + } + select { case <-done: return @@ -79,6 +93,20 @@ func waitForStateUpdates(t *testing.T, updateChan <-chan optimistic_sync.State, close(done) // make sure function exists after timeout } +// waitForStateUpdatesAndNoError behaves like waitForStateUpdates, but additionally requires that +// no error is queued in errChan once all expected states have been consumed. Use it at call sites +// that expect no error up to this point. Do NOT use it when an expected error may already be +// queued concurrently with the state updates (e.g. cancellation tests) - use waitForStateUpdates +// followed by waitForError instead. +func waitForStateUpdatesAndNoError(t *testing.T, updateChan <-chan optimistic_sync.State, errChan <-chan error, expectedStates ...optimistic_sync.State) { + waitForStateUpdates(t, updateChan, errChan, expectedStates...) + select { + case err := <-errChan: + require.NoError(t, err, "unexpected error queued behind state updates") + default: + } +} + // waitForErrorWithCustomCheckers waits for an error from the errChan within 500ms // and applies custom checker functions to validate the error. // If no checkers are provided, it asserts that no error occurred. diff --git a/module/util/log_test.go b/module/util/log_test.go index 6acc30b4441..35b940c5f5d 100644 --- a/module/util/log_test.go +++ b/module/util/log_test.go @@ -7,6 +7,7 @@ import ( "strings" "sync" "testing" + "testing/synctest" "time" "github.com/rs/zerolog" @@ -313,40 +314,46 @@ func TestLogProgressContinueLoggingAfter100(t *testing.T) { func TestLogProgressNoDataForAWhile(t *testing.T) { t.Parallel() - total := 1000 + // Run inside a synctest bubble with a fake clock: the "no data for a while" detection compares + // the wall-clock gap between two progress updates against the configured 1ms threshold. With a + // real clock, any scheduling hiccup between two updates yields an unexpected extra log line + // (flaky test). With the fake clock, time only advances during the explicit sleep below. + synctest.Test(t, func(t *testing.T) { + total := 1000 - buf := bytes.NewBufferString("") - lg := zerolog.New(buf) - logger := LogProgress( - lg, - NewLogProgressConfig[uint64]( - "test", - uint64(total), - 1*time.Millisecond, - 10, - ), - ) + buf := bytes.NewBufferString("") + lg := zerolog.New(buf) + logger := LogProgress( + lg, + NewLogProgressConfig[uint64]( + "test", + uint64(total), + 1*time.Millisecond, + 10, + ), + ) - for i := range total { - // somewhere in the middle pause for a bit - if i == 13 { - <-time.After(3 * time.Millisecond) - } + for i := range total { + // somewhere in the middle pause for a bit + if i == 13 { + time.Sleep(3 * time.Millisecond) + } - logger(1) - } + logger(1) + } - expectedLogs := []string{ - fmt.Sprintf(`test progress 0/%d`, total), - fmt.Sprintf(`test progress %d/%d (100.0%%)`, total, total), - } + expectedLogs := []string{ + fmt.Sprintf(`test progress 0/%d`, total), + fmt.Sprintf(`test progress %d/%d (100.0%%)`, total, total), + } - for _, log := range expectedLogs { - require.Contains(t, buf.String(), log, total) - } - lines := strings.Count(buf.String(), "\n") - // every 10% + 1 for the final log + 1 for the "no data in a while" log - require.Equal(t, 12, lines) + for _, log := range expectedLogs { + require.Contains(t, buf.String(), log, total) + } + lines := strings.Count(buf.String(), "\n") + // every 10% + 1 for the final log + 1 for the "no data in a while" log + require.Equal(t, 12, lines) + }) } func TestLogProgressMultipleGoroutines(t *testing.T) { diff --git a/network/p2p/builder/libp2pscaler_test.go b/network/p2p/builder/libp2pscaler_test.go index 4fd853dbd1c..245e71ae25f 100644 --- a/network/p2p/builder/libp2pscaler_test.go +++ b/network/p2p/builder/libp2pscaler_test.go @@ -1,6 +1,7 @@ package p2pbuilder import ( + "math/rand" "testing" "github.com/libp2p/go-libp2p" @@ -166,12 +167,36 @@ func TestBuildLibp2pResourceManagerLimits(t *testing.T) { defaultConcreteLimits, err := BuildLibp2pResourceManagerLimits(unittest.Logger(), &cfg.NetworkConfig.ResourceManager) require.NoError(t, err) + // overrideFixture generates a random override for a scope, such that every value is guaranteed to + // (i) be a positive concrete limit, avoiding the special rcmgr values (default = 0, unlimited = -1, + // block-all = -2), and (ii) differ from the corresponding default limit, so that the test can + // verify below that each value was indeed overridden. Fully random values (previously + // `unittest.LibP2PResourceLimitOverrideFixture`) occasionally collided with a default limit, + // making the test flaky. + overrideFixture := func(defaults rcmgr.ResourceLimits) p2pconfig.ResourceManagerOverrideLimit { + randomAbove := func(limit int) int { + if limit < 0 { + limit = 0 + } + return limit + 1 + rand.Intn(1000) + } + return p2pconfig.ResourceManagerOverrideLimit{ + StreamsInbound: randomAbove(int(defaults.StreamsInbound)), + StreamsOutbound: randomAbove(int(defaults.StreamsOutbound)), + ConnectionsInbound: randomAbove(int(defaults.ConnsInbound)), + ConnectionsOutbound: randomAbove(int(defaults.ConnsOutbound)), + FD: randomAbove(int(defaults.FD)), + Memory: randomAbove(int(defaults.Memory)), + } + } + // now the test creates random override configs for each scope, and re-build the concrete limits. - cfg.NetworkConfig.ResourceManager.Override.System = unittest.LibP2PResourceLimitOverrideFixture() - cfg.NetworkConfig.ResourceManager.Override.Transient = unittest.LibP2PResourceLimitOverrideFixture() - cfg.NetworkConfig.ResourceManager.Override.Protocol = unittest.LibP2PResourceLimitOverrideFixture() - cfg.NetworkConfig.ResourceManager.Override.Peer = unittest.LibP2PResourceLimitOverrideFixture() - cfg.NetworkConfig.ResourceManager.Override.PeerProtocol = unittest.LibP2PResourceLimitOverrideFixture() + defaultPartial := defaultConcreteLimits.ToPartialLimitConfig() + cfg.NetworkConfig.ResourceManager.Override.System = overrideFixture(defaultPartial.System) + cfg.NetworkConfig.ResourceManager.Override.Transient = overrideFixture(defaultPartial.Transient) + cfg.NetworkConfig.ResourceManager.Override.Protocol = overrideFixture(defaultPartial.ProtocolDefault) + cfg.NetworkConfig.ResourceManager.Override.Peer = overrideFixture(defaultPartial.PeerDefault) + cfg.NetworkConfig.ResourceManager.Override.PeerProtocol = overrideFixture(defaultPartial.ProtocolPeerDefault) overriddenConcreteLimits, err := BuildLibp2pResourceManagerLimits(unittest.Logger(), &cfg.NetworkConfig.ResourceManager) require.NoError(t, err) diff --git a/network/p2p/inspector/validation/control_message_validation_inspector_test.go b/network/p2p/inspector/validation/control_message_validation_inspector_test.go index 8258f598f0f..be17409411b 100644 --- a/network/p2p/inspector/validation/control_message_validation_inspector_test.go +++ b/network/p2p/inspector/validation/control_message_validation_inspector_test.go @@ -1431,6 +1431,11 @@ func TestNewControlMsgValidationInspector_validateClusterPrefixedTopic(t *testin // the 11th unknown cluster ID error should cause an error params.Config.ClusterPrefixedMessage.HardThreshold = 10 params.Config.GraftPrune.InvalidTopicIdThreshold = 0 + // Process the inspections sequentially with a single worker. With multiple workers, the + // cluster-prefix tracker increment and the subsequent hard-threshold check interleave, so + // more than one worker can observe the tracker above the hard threshold and disseminate + // more than one notification, violating the `Once()` expectation below. + params.Config.InspectionQueue.NumberOfWorkers = 1 params.Logger = logger }) clusterID := flow.ChainID(unittest.IdentifierFixture().String()) diff --git a/network/test/cohort1/network_test.go b/network/test/cohort1/network_test.go index 70b926c362e..bd742d35779 100644 --- a/network/test/cohort1/network_test.go +++ b/network/test/cohort1/network_test.go @@ -423,7 +423,6 @@ func (suite *NetworkTestSuite) TestUnicastRateLimit_Bandwidth() { require.Equal(suite.T(), reason, ratelimit.ReasonBandwidth.String()) // we only expect messages from the first network on the test suite - require.NoError(suite.T(), err) require.Equal(suite.T(), expectedPID, peerID) // update hook calls rateLimits.Inc() @@ -541,8 +540,12 @@ func (suite *NetworkTestSuite) TestUnicastRateLimit_Bandwidth() { // wait for all rate limits before shutting down network unittest.RequireCloseBefore(suite.T(), ch, 100*time.Millisecond, "could not stop on rate limit test ch on time") - // remote node should have received the first 2 messages - assert.Equal(suite.T(), uint64(2), callCount.Value()) + // remote node should have received the first 2 messages. Delivery to the receiving engine is + // asynchronous, so the messages may still be in flight when the sender observes the rate limit + // of the third message - hence, we wait instead of asserting immediately. + require.Eventually(suite.T(), func() bool { + return callCount.Value() == uint64(2) + }, time.Second, 10*time.Millisecond, "expected remote node to receive the first 2 messages") // sleep for 1 seconds to allow connection pruner to prune connections time.Sleep(1 * time.Second) @@ -551,6 +554,12 @@ func (suite *NetworkTestSuite) TestUnicastRateLimit_Bandwidth() { p2ptest.EnsureNotConnectedBetweenGroups(suite.T(), ctx, []p2p.LibP2PNode{libP2PNodes[0]}, []p2p.LibP2PNode{suite.libP2PNodes[0]}) p2pfixtures.EnsureNoStreamCreationBetweenGroups(suite.T(), ctx, []p2p.LibP2PNode{libP2PNodes[0]}, []p2p.LibP2PNode{suite.libP2PNodes[0]}) + // the rate-limited third message must never be delivered: after the prune window and with the + // connection confirmed pruned, the call count must still be exactly 2 (the Eventually above + // only checks that 2 is reached) + require.Equal(suite.T(), uint64(2), callCount.Value(), + "no messages beyond the first 2 may be delivered after the rate limit") + // eventually the rate limited node should be able to reconnect and send messages require.Eventually(suite.T(), func() bool { err = con0.Unicast(&libp2pmessage.TestMessage{ diff --git a/state/cluster/badger/mutator_test.go b/state/cluster/badger/mutator_test.go index cbf68967be3..7e4003636f1 100644 --- a/state/cluster/badger/mutator_test.go +++ b/state/cluster/badger/mutator_test.go @@ -490,12 +490,15 @@ func (suite *MutatorSuite) TestExtend_WithUnfinalizedReferenceBlock() { // to only use finalized blocks as reference, the proposer knowingly generated an invalid func (suite *MutatorSuite) TestExtend_WithOrphanedReferenceBlock() { // create a block extending genesis which is not finalized - orphaned := unittest.BlockWithParentProtocolState(suite.protoGenesis) + // Since `orphaned` and `finalized` are siblings, they must have different views. We track the views + // used so far in `usedViews` to prevent the fixtures from generating two blocks with the same view. + usedViews := make(map[uint64]struct{}) + orphaned := unittest.BlockWithParentProtocolStateAndUniqueView(suite.protoGenesis, usedViews) err := suite.protoState.ExtendCertified(context.Background(), unittest.NewCertifiedBlock(orphaned)) suite.Require().NoError(err) // create a block extending genesis (conflicting with previous) which is finalized - finalized := unittest.BlockWithParentProtocolState(suite.protoGenesis) + finalized := unittest.BlockWithParentProtocolStateAndUniqueView(suite.protoGenesis, usedViews) finalized.Payload.Guarantees = nil err = suite.protoState.ExtendCertified(context.Background(), unittest.NewCertifiedBlock(finalized)) suite.Require().NoError(err) diff --git a/storage/migration/validation_test.go b/storage/migration/validation_test.go index f7abf578586..5e683dd209f 100644 --- a/storage/migration/validation_test.go +++ b/storage/migration/validation_test.go @@ -117,18 +117,22 @@ func TestCompareKeyValuePairsFromChannels(t *testing.T) { badgerCh := make(chan KVPairs, len(tc.badgerKVs)) pebbleCh := make(chan KVPairs, len(tc.pebbleKVs)) - for _, kv := range tc.badgerKVs { - badgerCh <- kv - } - close(badgerCh) - for _, kv := range tc.pebbleKVs { - pebbleCh <- kv - } - close(pebbleCh) - if tc.name == "context cancelled" { - // Cancel context before running + // Cancel the context before running. We deliberately leave both channels empty and + // open: if channel reads were ready at the same time as the cancelled context, + // `select` would pick one of the ready cases at random, and the function could + // return successfully without ever observing the cancellation (flaky test). + // With empty, open channels, observing the cancellation is the only possibility. cancel() + } else { + for _, kv := range tc.badgerKVs { + badgerCh <- kv + } + close(badgerCh) + for _, kv := range tc.pebbleKVs { + pebbleCh <- kv + } + close(pebbleCh) } err := compareKeyValuePairsFromChannels(ctx, badgerCh, pebbleCh)