fix: the latest cached height could be zero (#98)

app/app.go

@@ -45,7 +45,7 @@ func NewApp(cfg *config.Config) *App {
 	db, err := gorm.Open(mysql.Open(dbPath), &gorm.Config{})
 
 	// only for debug purpose
-	//db = db.Debug()
+	db = db.Debug()
 
 	if err != nil {
 		panic(fmt.Sprintf("open db error, err=%+v", err.Error()))

executor/executor.go

@@ -146,11 +146,20 @@ func (e *Executor) GetLatestBlockHeight() (uint64, error) {
 	return latestHeight, nil
 }
 
-func (e *Executor) GetCachedBlockHeight() uint64 {
+func (e *Executor) GetCachedBlockHeight() (uint64, error) {
 	e.mtx.Lock()
 	cachedHeight := e.height
 	e.mtx.Unlock()
-	return cachedHeight
+
+	var err error
+	if cachedHeight == 0 {
+		cachedHeight, err = e.GetLatestBlockHeight()
+		if err != nil {
+			return 0, err
+		}
+	}
+
+	return cachedHeight, nil
 }
 
 func (e *Executor) queryLatestValidators() ([]*tmtypes.Validator, error) {

monitor/monitor.go

@@ -181,7 +181,7 @@ func (m *Monitor) calNextHeight() (uint64, error) {
 	if latestPolledBlock.Height == 0 { // a fresh database
 		latestHeight, err := m.executor.GetLatestBlockHeight()
 		if err != nil {
-			return m.executor.GetCachedBlockHeight(), err
+			return m.executor.GetCachedBlockHeight()
 		}
 		return latestHeight, nil
 	}

submitter/tx_submitter.go

@@ -59,7 +59,11 @@ func (s *TxSubmitter) SubmitTransactionLoop() {
 		// Loop until submitter is inturn to submit
 		attestPeriodEnd := s.queryAttestPeriodLoop()
 		// Fetch events for submit
-		currentHeight := s.executor.GetCachedBlockHeight()
+		currentHeight, err := s.executor.GetCachedBlockHeight()
+		if err != nil {
+			logging.Logger.Errorf("failed to get cached block height, err=%s", err.Error())
+			continue
+		}
 		events, err := s.FetchEventsForSubmit(currentHeight)
 		if err != nil {
 			s.metricService.IncSubmitterErr(err)
@@ -70,6 +74,12 @@ func (s *TxSubmitter) SubmitTransactionLoop() {
 			time.Sleep(common.RetryInterval)
 			continue
 		}
+		fetchedEvents := []uint64{}
+		for _, v := range events {
+			fetchedEvents = append(fetchedEvents, v.ChallengeId)
+		}
+		logging.Logger.Infof("submitter fetched these events at block height: %d, %+v", currentHeight, fetchedEvents)
+
 		// Submit events
 		for _, event := range events {
 			// Submitter no longer in-turn
@@ -220,7 +230,11 @@ func (s *TxSubmitter) submitTransactionLoop(event *model.Event, attestPeriodEnd
 
 // preCheck checks if the event has expired.
 func (s *TxSubmitter) preCheck(event *model.Event) error {
-	currentHeight := s.executor.GetCachedBlockHeight()
+	currentHeight, err := s.executor.GetCachedBlockHeight()
+	if err != nil {
+		logging.Logger.Errorf("failed to get cached block height, err=%s", err.Error())
+		return err
+	}
 	if event.ExpiredHeight < currentHeight {
 		logging.Logger.Infof("submitter for challengeId has expired. expired height, current height", event.ChallengeId, event.ExpiredHeight, currentHeight)
 		return common.ErrEventExpired

verifier/hash_verifier.go

@@ -70,7 +70,11 @@ func (v *Verifier) VerifyHashLoop() {
 }
 func (v *Verifier) verifyHash() error {
 	// Read unprocessed event from db with lowest challengeId
-	currentHeight := v.executor.GetCachedBlockHeight()
+	currentHeight, err := v.executor.GetCachedBlockHeight()
+	if err != nil {
+		logging.Logger.Errorf("failed to get cached block height, err=%s", err.Error())
+		return err
+	}
 	events, err := v.dataProvider.FetchEventsForVerification(currentHeight)
 	if err != nil {
 		v.metricService.IncHashVerifierErr(err)
@@ -81,7 +85,7 @@ func (v *Verifier) verifyHash() error {
 	for _, v := range events {
 		fetchedEvents = append(fetchedEvents, v.ChallengeId)
 	}
-	logging.Logger.Infof("verifier fetched these events for verification: %+v", fetchedEvents)
+	logging.Logger.Infof("verifier fetched these events for verification at block height: %d, %+v", currentHeight, fetchedEvents)
 
 	if len(events) == 0 {
 		time.Sleep(common.RetryInterval)
@@ -137,7 +141,11 @@ func (v *Verifier) verifyForSingleEvent(event *model.Event) error {
 	var err error
 	startTime := time.Now()
 	logging.Logger.Infof("verifier started for challengeId: %d %s", event.ChallengeId, time.Now().Format("15:04:05.000000"))
-	currentHeight := v.executor.GetCachedBlockHeight()
+	currentHeight, err := v.executor.GetCachedBlockHeight()
+	if err != nil {
+		logging.Logger.Errorf("failed to get cached block height, err=%s", err.Error())
+		return err
+	}
 	if err = v.preCheck(event, currentHeight); err != nil {
 		return err
 	}

vote/vote_broadcaster.go

@@ -45,7 +45,11 @@ func NewVoteBroadcaster(cfg *config.Config, signer *VoteSigner,
 
 func (p *VoteBroadcaster) BroadcastVotesLoop() {
 	for {
-		currentHeight := p.executor.GetCachedBlockHeight()
+		currentHeight, err := p.executor.GetCachedBlockHeight()
+		if err != nil {
+			logging.Logger.Errorf("failed to get cached block height, err=%s", err.Error())
+			continue
+		}
 		events, heartbeatEventCount, err := p.dataProvider.FetchEventsForSelfVote(currentHeight)
 		if err != nil {
 			p.metricService.IncBroadcasterErr(err)
@@ -61,6 +65,11 @@ func (p *VoteBroadcaster) BroadcastVotesLoop() {
 				p.metricService.IncHeartbeatEvents()
 			}
 		}
+		fetchedEvents := []uint64{}
+		for _, v := range events {
+			fetchedEvents = append(fetchedEvents, v.ChallengeId)
+		}
+		logging.Logger.Infof("broadcaster fetched these events at block height: %d, %+v", currentHeight, fetchedEvents)
 
 		for _, event := range events {
 			localVote, found := p.cachedLocalVote.Get(event.ChallengeId)
@@ -85,7 +94,6 @@ func (p *VoteBroadcaster) BroadcastVotesLoop() {
 
 			err = p.broadcastForSingleEvent(localVote.(*votepool.Vote), event)
 			if err != nil {
-				p.metricService.IncBroadcasterErr(err)
 				continue
 			}
 			time.Sleep(50 * time.Millisecond)
@@ -120,7 +128,11 @@ func (p *VoteBroadcaster) broadcastForSingleEvent(localVote *votepool.Vote, even
 }
 
 func (p *VoteBroadcaster) preCheck(event *model.Event) error {
-	currentHeight := p.executor.GetCachedBlockHeight()
+	currentHeight, err := p.executor.GetCachedBlockHeight()
+	if err != nil {
+		logging.Logger.Errorf("failed to get cached block height, err=%s", err.Error())
+		return err
+	}
 	if currentHeight > event.ExpiredHeight {
 		logging.Logger.Infof("broadcaster for challengeId: %d has expired. expired height: %d, current height: %d, timestamp: %s", event.ChallengeId, event.ExpiredHeight, currentHeight, time.Now().Format("15:04:05.000000"))
 		return common.ErrEventExpired

vote/vote_collator.go

@@ -38,15 +38,23 @@ func NewVoteCollator(cfg *config.Config, signer *VoteSigner,
 
 func (p *VoteCollator) CollateVotesLoop() {
 	for {
-		currentHeight := p.executor.GetCachedBlockHeight()
+		currentHeight, err := p.executor.GetCachedBlockHeight()
+		if err != nil {
+			logging.Logger.Errorf("failed to get cached block height, err=%s", err.Error())
+			continue
+		}
 		events, err := p.dataProvider.FetchEventsForCollate(currentHeight)
-		logging.Logger.Infof("vote processor fetched %d events for collate", len(events))
 		if err != nil {
 			p.metricService.IncCollatorErr(err)
 			logging.Logger.Errorf("vote processor failed to fetch unexpired events to collate votes, err=%+v", err.Error())
 			time.Sleep(RetryInterval)
 			continue
 		}
+		fetchedEvents := []uint64{}
+		for _, v := range events {
+			fetchedEvents = append(fetchedEvents, v.ChallengeId)
+		}
+		logging.Logger.Infof("collator fetched these events at block height: %d, %+v", currentHeight, fetchedEvents)
 		if len(events) == 0 {
 			time.Sleep(RetryInterval)
 			continue
@@ -107,7 +115,11 @@ func (p *VoteCollator) prepareEnoughValidVotesForEvent(event *model.Event) error
 }
 
 func (p *VoteCollator) preCheck(event *model.Event) error {
-	currentHeight := p.executor.GetCachedBlockHeight()
+	currentHeight, err := p.executor.GetCachedBlockHeight()
+	if err != nil {
+		logging.Logger.Errorf("failed to get cached block height, err=%s", err.Error())
+		return err
+	}
 	if currentHeight > event.ExpiredHeight {
 		logging.Logger.Infof("collator for challengeId: %d has expired. expired height: %d, current height: %d, timestamp: %s", event.ChallengeId, event.ExpiredHeight, currentHeight, time.Now().Format("15:04:05.000000"))
 		return common.ErrEventExpired