Fixed infinity loop sync.

api/service/stagedstreamsync/staged_stream_sync.go

@@ -642,8 +642,9 @@ func (ss *StagedStreamSync) addConsensusLastMile(bc core.BlockChain, cs *consens
 			case errors.Is(err, core.ErrNotLastBlockInEpoch):
 			case err != nil:
 				return errors.Wrap(err, "failed to InsertChain")
+			default:
+				hashes = append(hashes, block.Header().Hash())
 			}
-			hashes = append(hashes, block.Header().Hash())
 		}
 		return nil
 	})

api/service/stagedstreamsync/syncing.go

@@ -219,6 +219,8 @@ func (s *StagedStreamSync) Debug(source string, msg interface{}) {
 // For each iteration, estimate the current block number, then fetch block & insert to blockchain
 func (s *StagedStreamSync) doSync(downloaderContext context.Context, initSync bool) (uint64, int, error) {
 
+	startedNumber := s.bc.CurrentBlock().NumberU64()
+
 	var totalInserted int
 
 	s.initSync = initSync
@@ -249,7 +251,7 @@ func (s *StagedStreamSync) doSync(downloaderContext context.Context, initSync bo
 	for {
 		ctx, cancel := context.WithCancel(downloaderContext)
 
-		n, err := s.doSyncCycle(ctx, initSync)
+		n, err := s.doSyncCycle(ctx)
 		if err != nil {
 			utils.Logger().Error().
 				Err(err).
@@ -281,6 +283,8 @@ func (s *StagedStreamSync) doSync(downloaderContext context.Context, initSync bo
 			Bool("isBeacon", s.isBeacon).
 			Uint32("shard", s.bc.ShardID()).
 			Int("blocks", totalInserted).
+			Uint64("startedNumber", startedNumber).
+			Uint64("currentNumber", s.bc.CurrentBlock().NumberU64()).
 			Msg(WrapStagedSyncMsg("sync cycle blocks inserted successfully"))
 	}
 
@@ -304,7 +308,7 @@ func (s *StagedStreamSync) doSync(downloaderContext context.Context, initSync bo
 	return estimatedHeight, totalInserted, nil
 }
 
-func (s *StagedStreamSync) doSyncCycle(ctx context.Context, initSync bool) (int, error) {
+func (s *StagedStreamSync) doSyncCycle(ctx context.Context) (int, error) {
 
 	// TODO: initSync=true means currentCycleNumber==0, so we can remove initSync
 

consensus/consensus.go

@@ -40,6 +40,10 @@ const (
 	AsyncProposal
 )
 
+type DownloadAsync interface {
+	DownloadAsync()
+}
+
 // Consensus is the main struct with all states and data related to consensus process.
 type Consensus struct {
 	Decider quorum.Decider
@@ -122,9 +126,7 @@ type Consensus struct {
 	// finalityCounter keep tracks of the finality time
 	finalityCounter atomic.Value //int64
 
-	dHelper interface {
-		DownloadAsync()
-	}
+	dHelper DownloadAsync
 
 	// Both flags only for initialization state.
 	start           bool
@@ -190,10 +192,10 @@ func (consensus *Consensus) BlocksSynchronized() {
 }
 
 // BlocksNotSynchronized lets the main loop know that block is not synchronized
-func (consensus *Consensus) BlocksNotSynchronized() {
+func (consensus *Consensus) BlocksNotSynchronized(reason string) {
 	consensus.mutex.Lock()
 	defer consensus.mutex.Unlock()
-	consensus.syncNotReadyChan()
+	consensus.syncNotReadyChan(reason)
 }
 
 // VdfSeedSize returns the number of VRFs for VDF computation

consensus/consensus_v2.go

@@ -359,11 +359,12 @@ func (consensus *Consensus) syncReadyChan() {
 	}
 }
 
-func (consensus *Consensus) syncNotReadyChan() {
-	consensus.getLogger().Info().Msg("[ConsensusMainLoop] syncNotReadyChan")
+func (consensus *Consensus) syncNotReadyChan(reason string) {
+	mode := consensus.current.Mode()
 	consensus.setBlockNum(consensus.Blockchain().CurrentHeader().Number().Uint64() + 1)
 	consensus.current.SetMode(Syncing)
-	consensus.getLogger().Info().Msg("[ConsensusMainLoop] Node is OUT OF SYNC")
+	consensus.getLogger().Info().Msgf("[ConsensusMainLoop] syncNotReadyChan, prev %s, reason %s", mode.String(), reason)
+	consensus.getLogger().Info().Msgf("[ConsensusMainLoop] Node is OUT OF SYNC, reason: %s", reason)
 	consensusSyncCounterVec.With(prometheus.Labels{"consensus": "out_of_sync"}).Inc()
 }
 

consensus/downloader.go

@@ -61,7 +61,7 @@ func (dh *downloadHelper) downloadStartedLoop(c *Consensus) {
 	for {
 		select {
 		case <-dh.startedCh:
-			c.BlocksNotSynchronized()
+			c.BlocksNotSynchronized("downloadStartedLoop")
 
 		case err := <-dh.startedSub.Err():
 			c.GetLogger().Info().Err(err).Msg("consensus download finished loop closed")

consensus/validator.go

@@ -65,7 +65,7 @@ func (consensus *Consensus) onAnnounce(msg *msg_pb.Message) {
 			_, err := consensus.ValidateNewBlock(recvMsg)
 			if err == nil {
 				consensus.GetLogger().Info().
-					Msg("[Announce] Block verified")
+					Msgf("[Announce] Block verified %d", recvMsg.BlockNum)
 			}
 		}()
 	}

core/blockchain_impl.go

@@ -1608,7 +1608,7 @@ func (bc *BlockChainImpl) insertChain(chain types.Blocks, verifyHeaders bool) (i
 
 		switch status {
 		case CanonStatTy:
-			logger.Info().Msg("Inserted new block")
+			logger.Info().Msgf("Inserted new block s: %d e: %d n:%d", block.ShardID(), block.Epoch().Uint64(), block.NumberU64())
 			coalescedLogs = append(coalescedLogs, logs...)
 			blockInsertTimer.UpdateSince(bstart)
 			events = append(events, ChainEvent{block, block.Hash(), logs})

core/epochchain.go

@@ -166,7 +166,8 @@ func (bc *EpochChain) InsertChain(blocks types.Blocks, _ bool) (int, error) {
 		se1()
 		se2()
 		utils.Logger().Info().
-			Msgf("[EPOCHSYNC] Added block %d %s", block.NumberU64(), block.Hash().Hex())
+			Msgf("[EPOCHSYNC] Added block %d, epoch %d, %s", block.NumberU64(), block.Epoch().Uint64(), block.Hash().Hex())
+
 	}
 	return 0, nil
 }

core/rawdb/accessors_offchain.go

@@ -43,7 +43,7 @@ func WriteShardStateBytes(db DatabaseWriter, epoch *big.Int, data []byte) error
 	}
 	utils.Logger().Info().
 		Str("epoch", epoch.String()).
-		Int("size", len(data)).Msg("wrote sharding state")
+		Int("size", len(data)).Msgf("wrote sharding state, epoch %d", epoch.Uint64())
 	return nil
 }
 

node/node.go

@@ -499,10 +499,6 @@ func (node *Node) validateNodeMessage(ctx context.Context, payload []byte) (
 					utils.Logger().Debug().Uint64("receivedNum", block.NumberU64()).
 						Uint64("currentNum", curBeaconHeight).Msg("beacon block sync message rejected")
 					return nil, 0, errors.New("beacon block height smaller than current height beyond tolerance")
-				} else if block.NumberU64()-beaconBlockHeightTolerance > curBeaconHeight {
-					utils.Logger().Debug().Uint64("receivedNum", block.NumberU64()).
-						Uint64("currentNum", curBeaconHeight).Msg("beacon block sync message rejected")
-					return nil, 0, errors.Errorf("beacon block height too much higher than current height beyond tolerance, block %d, current %d, epoch %d , current %d", block.NumberU64(), curBeaconHeight, block.Epoch().Uint64(), curBeaconBlock.Epoch().Uint64())
 				} else if block.NumberU64() <= curBeaconHeight {
 					utils.Logger().Debug().Uint64("receivedNum", block.NumberU64()).
 						Uint64("currentNum", curBeaconHeight).Msg("beacon block sync message ignored")

node/node_handler.go

@@ -76,8 +76,6 @@ func (node *Node) HandleNodeMessage(
 				if node.Blockchain().ShardID() != shard.BeaconChainShardID {
 					for _, block := range blocks {
 						if block.ShardID() == 0 {
-							utils.Logger().Info().
-								Msgf("Beacon block being handled by block channel: %d", block.NumberU64())
 							if block.IsLastBlockInEpoch() {
 								go func(blk *types.Block) {
 									node.BeaconBlockChannel <- blk
@@ -339,7 +337,7 @@ func (node *Node) PostConsensusProcessing(newBlock *types.Block) error {
 		}
 		BroadcastCXReceipts(newBlock, node.Consensus)
 	} else {
-		if node.Consensus.Mode() != consensus.Listening {
+		if mode := node.Consensus.Mode(); mode != consensus.Listening {
 			numSignatures := node.Consensus.NumSignaturesIncludedInBlock(newBlock)
 			utils.Logger().Info().
 				Uint64("blockNum", newBlock.NumberU64()).
@@ -349,9 +347,12 @@ func (node *Node) PostConsensusProcessing(newBlock *types.Block) error {
 				Int("numTxns", len(newBlock.Transactions())).
 				Int("numStakingTxns", len(newBlock.StakingTransactions())).
 				Uint32("numSignatures", numSignatures).
+				Str("mode", mode.String()).
 				Msg("BINGO !!! Reached Consensus")
 			if node.Consensus.Mode() == consensus.Syncing {
-				node.Consensus.SetMode(node.Consensus.UpdateConsensusInformation())
+				mode = node.Consensus.UpdateConsensusInformation()
+				utils.Logger().Info().Msgf("Switching to mode %s", mode)
+				node.Consensus.SetMode(mode)
 			}
 
 			node.Consensus.UpdateValidatorMetrics(float64(numSignatures), float64(newBlock.NumberU64()))

node/node_syncing.go

@@ -231,7 +231,16 @@ func (node *Node) doBeaconSyncing() {
 		// If Downloader is not working, we need also deal with blocks from beaconBlockChannel
 		go func(node *Node) {
 			// TODO ek – infinite loop; add shutdown/cleanup logic
-			for _ = range node.BeaconBlockChannel {
+			for b := range node.BeaconBlockChannel {
+				if b != nil && b.IsLastBlockInEpoch() {
+					_, err := node.EpochChain().InsertChain(types.Blocks{b}, true)
+					if err != nil {
+						utils.Logger().Error().Err(err).Msgf("[SYNC] InsertChain failed shard: %d epoch:%d number:%d", b.Header().ShardID(), b.Epoch().Uint64(), b.NumberU64())
+					} else {
+						utils.Logger().Info().
+							Msgf("Beacon block being handled by block channel: epoch: %d, number: %d", b.Epoch().Uint64(), b.NumberU64())
+					}
+				}
 			}
 		}(node)
 	}
@@ -307,7 +316,7 @@ func (node *Node) doSync(syncInstance ISync, syncingPeerProvider SyncingPeerProv
 	if isSynchronized, _, _ := syncInstance.GetParsedSyncStatusDoubleChecked(); !isSynchronized {
 		node.IsSynchronized.UnSet()
 		if willJoinConsensus {
-			consensus.BlocksNotSynchronized()
+			consensus.BlocksNotSynchronized("node.doSync")
 		}
 		isBeacon := bc.ShardID() == shard.BeaconChainShardID
 		syncInstance.SyncLoop(bc, isBeacon, consensus, legacysync.LoopMinTime)