Merge #4134

4134: Re-enable SN epoch integration test r=jordanschalm a=jordanschalm

## Context

The test failure was caused by this fatal error from the Consensus Node brought up to join in the second epoch:
`··· DOCK: 2023-03-30 19:48:34.635587 +0000 UTC (clogs ) consensus_test_3          (16726f0fad5781fa651f6a048c007e9f6a4202f8111811cbec685ac96774c32a) - �{"level":"fatal","node_role":"consensus","node_id":"522f1406e6e0c2267dbcc49519539c18d866a584df709e1a277bdbc28bd2d40c","error":"component sealing engine initialization failed: could not repopulate assignment collectors tree: internal error while traversing fork: visitor errored on block 6058fc1a85eaff3f46fe1d2a530ea7fe648cef771c75594bcfeef66238450bda at height 57: could not process incorporated result from block 6058fc1a85eaff3f46fe1d2a530ea7fe648cef771c75594bcfeef66238450bda: could not process incorporated incRes: could not process incorporated result 741f4ba4ad5ff7918b7aaa615306f136ae2e2b9b13a729e115e9fb48c2713b03: could not determine chunk assignment: failed to retrieve source of randomness: could not retrieve quorum certificate for (6058fc1a85eaff3f46fe1d2a530ea7fe648cef771c75594bcfeef66238450bda): could not retrieve resource: key not found","time":"2023-03-30T19:48:34.634479451Z","message":"unhandled irrecoverable error"}`

The problem is that, the sealing engine attempts to populate its state using the sealing segment, but it cannot read the QC for some of these blocks. In #3947, we changed to read QCs from storage directly in all cases. However, these QCs were not stored in QC storage for all blocks, in particular sealing segment blocks. 

## Changes
- Fixes the bug by inserting QCs explicitly for all sealing segment blocks and adds a test case
- Re-enables the integration test

Co-authored-by: Jordan Schalm <jordan@dapperlabs.com>

integration/tests/epochs/epoch_join_and_leave_sn_test.go

@@ -6,7 +6,6 @@ import (
 	"github.com/stretchr/testify/suite"
 
 	"github.com/onflow/flow-go/model/flow"
-	"github.com/onflow/flow-go/utils/unittest"
 )
 
 func TestEpochJoinAndLeaveSN(t *testing.T) {
@@ -20,6 +19,5 @@ type EpochJoinAndLeaveSNSuite struct {
 // TestEpochJoinAndLeaveSN should update consensus nodes and assert healthy network conditions
 // after the epoch transition completes. See health check function for details.
 func (s *EpochJoinAndLeaveSNSuite) TestEpochJoinAndLeaveSN() {
-	unittest.SkipUnless(s.T(), unittest.TEST_FLAKY, "fails on CI regularly")
 	s.runTestEpochJoinAndLeave(flow.RoleConsensus, s.assertNetworkHealthyAfterSNChange)
 }

integration/tests/epochs/suite.go

@@ -672,7 +672,7 @@ func (s *Suite) runTestEpochJoinAndLeave(role flow.Role, checkNetworkHealth node
 	s.TimedLogf("retrieved header after entering EpochSetup phase: root_height=%d, root_view=%d, segment_heights=[%d-%d], segment_views=[%d-%d]",
 		header.Height, header.View,
 		segment.Sealed().Header.Height, segment.Highest().Header.Height,
-		segment.Sealed().Header.View, segment.Highest().Header.Height)
+		segment.Sealed().Header.View, segment.Highest().Header.View)
 
 	testContainer.WriteRootSnapshot(rootSnapshot)
 	testContainer.Container.Start(s.ctx)

state/protocol/badger/state.go

@@ -194,11 +194,15 @@ func (state *State) bootstrapSealingSegment(segment *flow.SealingSegment, head *
 			height := block.Header.Height
 			err := state.blocks.StoreTx(block)(tx)
 			if err != nil {
-				return fmt.Errorf("could not insert root block: %w", err)
+				return fmt.Errorf("could not insert SealingSegment extra block: %w", err)
 			}
 			err = transaction.WithTx(operation.IndexBlockHeight(height, blockID))(tx)
 			if err != nil {
-				return fmt.Errorf("could not index root block segment (id=%x): %w", blockID, err)
+				return fmt.Errorf("could not index SealingSegment extra block (id=%x): %w", blockID, err)
+			}
+			err = state.qcs.StoreTx(block.Header.QuorumCertificate())(tx)
+			if err != nil {
+				return fmt.Errorf("could not store qc for SealingSegment extra block (id=%x): %w", blockID, err)
 			}
 		}
 
@@ -208,11 +212,15 @@ func (state *State) bootstrapSealingSegment(segment *flow.SealingSegment, head *
 
 			err := state.blocks.StoreTx(block)(tx)
 			if err != nil {
-				return fmt.Errorf("could not insert root block: %w", err)
+				return fmt.Errorf("could not insert SealingSegment block: %w", err)
 			}
 			err = transaction.WithTx(operation.IndexBlockHeight(height, blockID))(tx)
 			if err != nil {
-				return fmt.Errorf("could not index root block segment (id=%x): %w", blockID, err)
+				return fmt.Errorf("could not index SealingSegment block (id=%x): %w", blockID, err)
+			}
+			err = state.qcs.StoreTx(block.Header.QuorumCertificate())(tx)
+			if err != nil {
+				return fmt.Errorf("could not store qc for SealingSegment block (id=%x): %w", blockID, err)
 			}
 
 			// index the latest seal as of this block

state/protocol/badger/state_test.go

@@ -280,6 +280,16 @@ func TestBootstrapNonRoot(t *testing.T) {
 		bootstrap(t, after, func(state *bprotocol.State, err error) {
 			require.NoError(t, err)
 			unittest.AssertSnapshotsEqual(t, after, state.Final())
+			// should be able to read all QCs
+			segment, err := state.Final().SealingSegment()
+			require.NoError(t, err)
+			for _, block := range segment.Blocks {
+				snapshot := state.AtBlockID(block.ID())
+				_, err := snapshot.QuorumCertificate()
+				require.NoError(t, err)
+				_, err = snapshot.RandomSource()
+				require.NoError(t, err)
+			}
 		})
 	})