Process missed blocks on startup

database/database.go

@@ -9,7 +9,7 @@ import (
 )
 
 const (
-	LatestSeenBlockKey = "latestSeenBlock"
+	LatestProcessedBlockKey = "latestProcessedBlock"
 )
 
 var (

database/json_file_storage_test.go

@@ -38,15 +38,15 @@ func TestConcurrentWriteReadSingleChain(t *testing.T) {
 	finalTargetValue := uint64(11)
 	testWrite(jsonStorage, networks[0], finalTargetValue)
 
-	latestSeenBlockData, err := jsonStorage.Get(networks[0], []byte(LatestSeenBlockKey))
+	latestProcessedBlockData, err := jsonStorage.Get(networks[0], []byte(LatestProcessedBlockKey))
 	if err != nil {
 		t.Fatalf("failed to retrieve from JSON storage. err: %v", err)
 	}
-	latestSeenBlock, success := new(big.Int).SetString(string(latestSeenBlockData), 10)
+	latestProcessedBlock, success := new(big.Int).SetString(string(latestProcessedBlockData), 10)
 	if !success {
 		t.Fatalf("failed to convert latest block to big.Int. err: %v", err)
 	}
-	assert.Equal(t, finalTargetValue, latestSeenBlock.Uint64(), "latest seen block height is not correct.")
+	assert.Equal(t, finalTargetValue, latestProcessedBlock.Uint64(), "latest processed block height is not correct.")
 
 }
 
@@ -78,15 +78,15 @@ func TestConcurrentWriteReadMultipleChains(t *testing.T) {
 	}
 
 	for i, id := range networks {
-		latestSeenBlockData, err := jsonStorage.Get(id, []byte(LatestSeenBlockKey))
+		latestProcessedBlockData, err := jsonStorage.Get(id, []byte(LatestProcessedBlockKey))
 		if err != nil {
 			t.Fatalf("failed to retrieve from JSON storage. networkID: %d err: %v", i, err)
 		}
-		latestSeenBlock, success := new(big.Int).SetString(string(latestSeenBlockData), 10)
+		latestProcessedBlock, success := new(big.Int).SetString(string(latestProcessedBlockData), 10)
 		if !success {
 			t.Fatalf("failed to convert latest block to big.Int. err: %v", err)
 		}
-		assert.Equal(t, finalTargetValue, latestSeenBlock.Uint64(), fmt.Sprintf("latest seen block height is not correct. networkID: %d", i))
+		assert.Equal(t, finalTargetValue, latestProcessedBlock.Uint64(), fmt.Sprintf("latest processed block height is not correct. networkID: %d", i))
 	}
 }
 
@@ -110,7 +110,7 @@ func setupJsonStorage(t *testing.T, networks []ids.ID) *JSONFileStorage {
 
 func testWrite(storage *JSONFileStorage, chainID ids.ID, height uint64) {
 	fmt.Println(chainID, height)
-	err := storage.Put(chainID, []byte(LatestSeenBlockKey), []byte(strconv.FormatUint(height, 10)))
+	err := storage.Put(chainID, []byte(LatestProcessedBlockKey), []byte(strconv.FormatUint(height, 10)))
 	if err != nil {
 		fmt.Printf("failed to put data: %v", err)
 		return

relayer/relayer.go

@@ -116,26 +116,26 @@ func NewRelayer(
 		return nil, nil, err
 	}
 
-	// Get the latest seen block height from the database.
-	latestSeenBlockData, err := r.db.Get(r.sourceChainID, []byte(database.LatestSeenBlockKey))
+	// Get the latest processed block height from the database.
+	latestProcessedBlockData, err := r.db.Get(r.sourceChainID, []byte(database.LatestProcessedBlockKey))
 
 	// The following cases are treated as successful:
-	// 1) The database contains the latest seen block data for the chain
+	// 1) The database contains the latest processed block data for the chain
 	//    - In this case, we parse the block height and process warp logs from that height to the current block
-	// 2) The database has been configured for the chain, but does not contain the latest seen block data
+	// 2) The database has been configured for the chain, but does not contain the latest processed block data
 	//    - In this case, we save the current block height in the database, but do not process any historical warp logs
 	if err == nil {
-		// If the database contains the latest seen block data, then back-process all warp messages from that block to the latest block
-		// Note that the latest seen block may have already been partially (or fully) processed by the relayer on a previous run. When
-		// processing a warp message in real time, which is when we update the latest seen block in the database, we have no way of knowing
+		// If the database contains the latest processed block data, then back-process all warp messages from that block to the latest block
+		// Note that the retrieved latest processed block may have already been partially (or fully) processed by the relayer on a previous run. When
+		// processing a warp message in real time, which is when we update the latest processed block in the database, we have no way of knowing
 		// if that is the last warp message in the block
-		latestSeenBlock, success := new(big.Int).SetString(string(latestSeenBlockData), 10)
+		latestProcessedBlock, success := new(big.Int).SetString(string(latestProcessedBlockData), 10)
 		if !success {
 			r.logger.Error("failed to convert latest block to big.Int", zap.Error(err))
 			return nil, nil, err
 		}
 
-		err = sub.ProcessFromHeight(latestSeenBlock)
+		err = sub.ProcessFromHeight(latestProcessedBlock)
 		if err != nil {
 			logger.Warn(
 				"Encountered an error when processing historical blocks. Continuing to normal relaying operation.",
@@ -146,16 +146,16 @@ func NewRelayer(
 		return &r, sub, nil
 	}
 	if errors.Is(err, database.ErrChainNotFound) || errors.Is(err, database.ErrKeyNotFound) {
-		// Otherwise, latestSeenBlock is nil, so we instead store the latest block height.
+		// Otherwise, latestProcessedBlock is nil, so we instead store the latest block height.
 		logger.Info(
-			"Latest seen block not found in database. Starting from latest block.",
+			"Latest processed block not found in database. Starting from latest block.",
 			zap.String("chainID", r.sourceChainID.String()),
 		)
 
-		err := sub.UpdateLatestSeenBlock()
+		err := sub.UpdateLatestProcessedBlock()
 		if err != nil {
 			logger.Warn(
-				"Failed to update latest seen block. Continuing to normal relaying operation",
+				"Failed to update latest processed block. Continuing to normal relaying operation",
 				zap.String("chainID", r.sourceChainID.String()),
 				zap.Error(err),
 			)
@@ -234,11 +234,11 @@ func (r *Relayer) RelayMessage(warpLogInfo *vmtypes.WarpLogInfo, metrics *Messag
 	// Increment the request ID for the next message relay request
 	r.currentRequestID++
 
-	// Update the database with the latest seen block height
-	err = r.db.Put(r.sourceChainID, []byte(database.LatestSeenBlockKey), []byte(strconv.FormatUint(warpLogInfo.BlockNumber, 10)))
+	// Update the database with the latest processed block height
+	err = r.db.Put(r.sourceChainID, []byte(database.LatestProcessedBlockKey), []byte(strconv.FormatUint(warpLogInfo.BlockNumber, 10)))
 	if err != nil {
 		r.logger.Error(
-			fmt.Sprintf("failed to put %s into database", database.LatestSeenBlockKey),
+			fmt.Sprintf("failed to put %s into database", database.LatestProcessedBlockKey),
 			zap.Error(err),
 		)
 	}

vms/evm/subscriber.go

@@ -176,7 +176,7 @@ func (s *subscriber) ProcessFromHeight(height *big.Int) error {
 		height = big.NewInt(0).Add(toBlock, big.NewInt(-MaxBlocksToProcess))
 	}
 
-	// Filter logs from the latest seen block to the latest block
+	// Filter logs from the latest processed block to the latest block
 	// Since initializationFilterQuery does not modify existing fields of warpFilterQuery,
 	// we can safely reuse warpFilterQuery with only a shallow copy
 	initializationFilterQuery := interfaces.FilterQuery{
@@ -226,9 +226,9 @@ func (s *subscriber) ProcessFromHeight(height *big.Int) error {
 	return nil
 }
 
-func (s *subscriber) UpdateLatestSeenBlock() error {
+func (s *subscriber) UpdateLatestProcessedBlock() error {
 	s.logger.Info(
-		"Updating latest seen block in database",
+		"Updating latest processed block in database",
 		zap.String("chainID", s.chainID.String()),
 	)
 	ethClient, err := ethclient.Dial(s.nodeRPCURL)
@@ -251,10 +251,10 @@ func (s *subscriber) UpdateLatestSeenBlock() error {
 		return err
 	}
 
-	err = s.db.Put(s.chainID, []byte(database.LatestSeenBlockKey), []byte(strconv.FormatUint(latestBlock, 10)))
+	err = s.db.Put(s.chainID, []byte(database.LatestProcessedBlockKey), []byte(strconv.FormatUint(latestBlock, 10)))
 	if err != nil {
 		s.logger.Error(
-			fmt.Sprintf("failed to put %s into database", database.LatestSeenBlockKey),
+			fmt.Sprintf("failed to put %s into database", database.LatestProcessedBlockKey),
 			zap.String("chainID", s.chainID.String()),
 			zap.Error(err),
 		)

vms/subscriber.go

@@ -20,8 +20,8 @@ type Subscriber interface {
 	// ProcessFromHeight processes events from {height} to the latest block
 	ProcessFromHeight(height *big.Int) error
 
-	// UpdateLatestSeenBlock retrieves the latest block from the chain and updates the database
-	UpdateLatestSeenBlock() error
+	// UpdateLatestProcessedBlock retrieves the latest block from the chain and updates the database
+	UpdateLatestProcessedBlock() error
 
 	// Subscribe registers a subscription. After Subscribe is called,
 	// log events that match [filter] are written to the channel returned