feat(dot/sync): include block origin and skip extra validation on initialSync

dot/sync/chain_sync.go

@@ -37,6 +37,13 @@ const (
 	tip
 )
 
+type blockOrigin byte
+
+const (
+	networkInitialSync blockOrigin = iota
+	networkBroadcast
+)
+
 func (s chainSyncState) String() string {
 	switch s {
 	case bootstrap:
@@ -169,23 +176,48 @@ func newChainSync(cfg chainSyncConfig) *chainSync {
 	}
 }
 
+func (cs *chainSync) waitEnoughPeersAndTarget() <-chan struct{} {
+	resultCh := make(chan struct{})
+	go func() {
+		defer close(resultCh)
+		for {
+			select {
+			case <-resultCh:
+				return
+			default:
+			}
+
+			cs.workerPool.useConnectedPeers()
+			_, err := cs.getTarget()
+
+			totalAvailable := cs.workerPool.totalWorkers()
+			if totalAvailable >= uint(cs.minPeers) && err == nil {
+				return
+			}
+
+			// TODO: https://github.com/ChainSafe/gossamer/issues/3402
+			time.Sleep(100 * time.Millisecond)
+		}
+	}()
+
+	return resultCh
+}
+
 func (cs *chainSync) start() {
 	// since the default status from sync mode is syncMode(tip)
 	isSyncedGauge.Set(1)
 
+	cs.wg.Add(1)
+	go cs.pendingBlocks.run(cs.finalisedCh, cs.stopCh, &cs.wg)
+
 	// wait until we have a minimal workers in the sync worker pool
-	for {
-		totalAvailable := cs.workerPool.totalWorkers()
-		if totalAvailable >= uint(cs.minPeers) {
-			break
-		}
+	resultCh := cs.waitEnoughPeersAndTarget()
 
-		// TODO: https://github.com/ChainSafe/gossamer/issues/3402
-		time.Sleep(time.Second)
+	select {
+	case <-resultCh:
+	case <-cs.stopCh:
+		return
 	}
-
-	cs.wg.Add(1)
-	go cs.pendingBlocks.run(cs.finalisedCh, cs.stopCh, &cs.wg)
 }
 
 func (cs *chainSync) stop() error {
@@ -265,7 +297,7 @@ func (cs *chainSync) bootstrapSync() {
 
 		if isFarFromTarget {
 			cs.workerPool.useConnectedPeers()
-			err = cs.requestMaxBlocksFrom(bestBlockHeader)
+			err = cs.requestMaxBlocksFrom(bestBlockHeader, networkInitialSync)
 			if err != nil {
 				logger.Errorf("requesting max blocks from best block header: %s", err)
 			}
@@ -424,7 +456,7 @@ func (cs *chainSync) requestChainBlocks(announcedHeader, bestBlockHeader *types.
 
 	resultsQueue := make(chan *syncTaskResult)
 	cs.workerPool.submitRequest(request, &peerWhoAnnounced, resultsQueue)
-	err := cs.handleWorkersResults(resultsQueue, startAtBlock, totalBlocks)
+	err := cs.handleWorkersResults(resultsQueue, networkBroadcast, startAtBlock, totalBlocks)
 	if err != nil {
 		return fmt.Errorf("while handling workers results: %w", err)
 	}
@@ -462,7 +494,7 @@ func (cs *chainSync) requestForkBlocks(bestBlockHeader, highestFinalizedHeader,
 	resultsQueue := make(chan *syncTaskResult)
 	cs.workerPool.submitRequest(request, &peerWhoAnnounced, resultsQueue)
 
-	err = cs.handleWorkersResults(resultsQueue, startAtBlock, gapLength)
+	err = cs.handleWorkersResults(resultsQueue, networkBroadcast, startAtBlock, gapLength)
 	if err != nil {
 		return fmt.Errorf("while handling workers results: %w", err)
 	}
@@ -490,7 +522,7 @@ func (cs *chainSync) requestPendingBlocks(highestFinalizedHeader *types.Header)
 		}
 
 		if parentExists {
-			err := cs.handleReadyBlock(pendingBlock.toBlockData())
+			err := cs.handleReadyBlock(pendingBlock.toBlockData(), networkBroadcast)
 			if err != nil {
 				return fmt.Errorf("handling ready block: %w", err)
 			}
@@ -515,7 +547,7 @@ func (cs *chainSync) requestPendingBlocks(highestFinalizedHeader *types.Header)
 		// TODO: we should handle the requests concurrently
 		// a way of achieve that is by constructing a new `handleWorkersResults` for
 		// handling only tip sync requests
-		err = cs.handleWorkersResults(resultsQueue, startAtBlock, *descendingGapRequest.Max)
+		err = cs.handleWorkersResults(resultsQueue, networkBroadcast, startAtBlock, *descendingGapRequest.Max)
 		if err != nil {
 			return fmt.Errorf("while handling workers results: %w", err)
 		}
@@ -524,7 +556,7 @@ func (cs *chainSync) requestPendingBlocks(highestFinalizedHeader *types.Header)
 	return nil
 }
 
-func (cs *chainSync) requestMaxBlocksFrom(bestBlockHeader *types.Header) error {
+func (cs *chainSync) requestMaxBlocksFrom(bestBlockHeader *types.Header, origin blockOrigin) error { //nolint:unparam
 	startRequestAt := bestBlockHeader.Number + 1
 
 	// targetBlockNumber is the virtual target we will request, however
@@ -551,7 +583,7 @@ func (cs *chainSync) requestMaxBlocksFrom(bestBlockHeader *types.Header) error {
 	}
 
 	resultsQueue := cs.workerPool.submitRequests(requests)
-	err = cs.handleWorkersResults(resultsQueue, startRequestAt, expectedAmountOfBlocks)
+	err = cs.handleWorkersResults(resultsQueue, origin, startRequestAt, expectedAmountOfBlocks)
 	if err != nil {
 		return fmt.Errorf("while handling workers results: %w", err)
 	}
@@ -589,7 +621,7 @@ func (cs *chainSync) getTarget() (uint, error) {
 // in the queue and wait for it to completes
 // TODO: handle only justification requests
 func (cs *chainSync) handleWorkersResults(
-	workersResults chan *syncTaskResult, startAtBlock uint, expectedSyncedBlocks uint32) error {
+	workersResults chan *syncTaskResult, origin blockOrigin, startAtBlock uint, expectedSyncedBlocks uint32) error {
 
 	startTime := time.Now()
 	defer func() {
@@ -739,15 +771,15 @@ taskResultLoop:
 	// response was validated! place into ready block queue
 	for _, bd := range syncingChain {
 		// block is ready to be processed!
-		if err := cs.handleReadyBlock(bd); err != nil {
+		if err := cs.handleReadyBlock(bd, origin); err != nil {
 			return fmt.Errorf("while handling ready block: %w", err)
 		}
 	}
 
 	return nil
 }
 
-func (cs *chainSync) handleReadyBlock(bd *types.BlockData) error {
+func (cs *chainSync) handleReadyBlock(bd *types.BlockData, origin blockOrigin) error {
 	// if header was not requested, get it from the pending set
 	// if we're expecting headers, validate should ensure we have a header
 	if bd.Header == nil {
@@ -779,7 +811,7 @@ func (cs *chainSync) handleReadyBlock(bd *types.BlockData) error {
 		bd.Header = block.header
 	}
 
-	err := cs.processBlockData(*bd)
+	err := cs.processBlockData(*bd, origin)
 	if err != nil {
 		// depending on the error, we might want to save this block for later
 		logger.Errorf("block data processing for block with hash %s failed: %s", bd.Hash, err)
@@ -793,7 +825,7 @@ func (cs *chainSync) handleReadyBlock(bd *types.BlockData) error {
 // processBlockData processes the BlockData from a BlockResponse and
 // returns the index of the last BlockData it handled on success,
 // or the index of the block data that errored on failure.
-func (cs *chainSync) processBlockData(blockData types.BlockData) error {
+func (cs *chainSync) processBlockData(blockData types.BlockData, origin blockOrigin) error {
 	headerInState, err := cs.blockState.HasHeader(blockData.Hash)
 	if err != nil {
 		return fmt.Errorf("checking if block state has header: %w", err)
@@ -817,7 +849,7 @@ func (cs *chainSync) processBlockData(blockData types.BlockData) error {
 
 	if blockData.Header != nil {
 		if blockData.Body != nil {
-			err = cs.processBlockDataWithHeaderAndBody(blockData, announceImportedBlock)
+			err = cs.processBlockDataWithHeaderAndBody(blockData, origin, announceImportedBlock)
 			if err != nil {
 				return fmt.Errorf("processing block data with header and body: %w", err)
 			}
@@ -885,10 +917,13 @@ func (cs *chainSync) processBlockDataWithStateHeaderAndBody(blockData types.Bloc
 }
 
 func (cs *chainSync) processBlockDataWithHeaderAndBody(blockData types.BlockData,
-	announceImportedBlock bool) (err error) {
-	err = cs.babeVerifier.VerifyBlock(blockData.Header)
-	if err != nil {
-		return fmt.Errorf("babe verifying block: %w", err)
+	origin blockOrigin, announceImportedBlock bool) (err error) {
+
+	if origin != networkInitialSync {
+		err = cs.babeVerifier.VerifyBlock(blockData.Header)
+		if err != nil {
+			return fmt.Errorf("babe verifying block: %w", err)
+		}
 	}
 
 	cs.handleBody(blockData.Body)

dot/sync/chain_sync_test.go

@@ -553,7 +553,7 @@ func TestChainSync_BootstrapSync_SuccessfulSync_WithOneWorker(t *testing.T) {
 	// the worker pool executes the workers management
 	cs.workerPool.fromBlockAnnounce(peer.ID("noot"))
 
-	err = cs.requestMaxBlocksFrom(mockedGenesisHeader)
+	err = cs.requestMaxBlocksFrom(mockedGenesisHeader, networkInitialSync)
 	require.NoError(t, err)
 
 	err = cs.workerPool.stop()
@@ -639,7 +639,7 @@ func TestChainSync_BootstrapSync_SuccessfulSync_WithTwoWorkers(t *testing.T) {
 	cs.workerPool.fromBlockAnnounce(peer.ID("noot"))
 	cs.workerPool.fromBlockAnnounce(peer.ID("noot2"))
 
-	err = cs.requestMaxBlocksFrom(mockedGenesisHeader)
+	err = cs.requestMaxBlocksFrom(mockedGenesisHeader, networkInitialSync)
 	require.NoError(t, err)
 
 	err = cs.workerPool.stop()
@@ -733,7 +733,7 @@ func TestChainSync_BootstrapSync_SuccessfulSync_WithOneWorkerFailing(t *testing.
 	cs.workerPool.fromBlockAnnounce(peer.ID("alice"))
 	cs.workerPool.fromBlockAnnounce(peer.ID("bob"))
 
-	err = cs.requestMaxBlocksFrom(mockedGenesisHeader)
+	err = cs.requestMaxBlocksFrom(mockedGenesisHeader, networkInitialSync)
 	require.NoError(t, err)
 
 	err = cs.workerPool.stop()
@@ -833,7 +833,7 @@ func TestChainSync_BootstrapSync_SuccessfulSync_WithProtocolNotSupported(t *test
 	cs.workerPool.fromBlockAnnounce(peer.ID("alice"))
 	cs.workerPool.fromBlockAnnounce(peer.ID("bob"))
 
-	err = cs.requestMaxBlocksFrom(mockedGenesisHeader)
+	err = cs.requestMaxBlocksFrom(mockedGenesisHeader, networkInitialSync)
 	require.NoError(t, err)
 
 	err = cs.workerPool.stop()
@@ -935,7 +935,7 @@ func TestChainSync_BootstrapSync_SuccessfulSync_WithNilHeaderInResponse(t *testi
 	cs.workerPool.fromBlockAnnounce(peer.ID("alice"))
 	cs.workerPool.fromBlockAnnounce(peer.ID("bob"))
 
-	err = cs.requestMaxBlocksFrom(mockedGenesisHeader)
+	err = cs.requestMaxBlocksFrom(mockedGenesisHeader, networkInitialSync)
 	require.NoError(t, err)
 
 	err = cs.workerPool.stop()
@@ -1033,7 +1033,7 @@ func TestChainSync_BootstrapSync_SuccessfulSync_WithResponseIsNotAChain(t *testi
 	cs.workerPool.fromBlockAnnounce(peer.ID("alice"))
 	cs.workerPool.fromBlockAnnounce(peer.ID("bob"))
 
-	err = cs.requestMaxBlocksFrom(mockedGenesisHeader)
+	err = cs.requestMaxBlocksFrom(mockedGenesisHeader, networkInitialSync)
 	require.NoError(t, err)
 
 	err = cs.workerPool.stop()
@@ -1147,7 +1147,7 @@ func TestChainSync_BootstrapSync_SuccessfulSync_WithReceivedBadBlock(t *testing.
 	cs.workerPool.fromBlockAnnounce(peer.ID("alice"))
 	cs.workerPool.fromBlockAnnounce(peer.ID("bob"))
 
-	err = cs.requestMaxBlocksFrom(mockedGenesisHeader)
+	err = cs.requestMaxBlocksFrom(mockedGenesisHeader, networkInitialSync)
 	require.NoError(t, err)
 
 	err = cs.workerPool.stop()
@@ -1229,7 +1229,7 @@ func TestChainSync_BootstrapSync_SucessfulSync_ReceivedPartialBlockData(t *testi
 
 	cs.workerPool.fromBlockAnnounce(peer.ID("alice"))
 
-	err = cs.requestMaxBlocksFrom(mockedGenesisHeader)
+	err = cs.requestMaxBlocksFrom(mockedGenesisHeader, networkInitialSync)
 	require.NoError(t, err)
 
 	err = cs.workerPool.stop()