Merge branch 'ryan/fix-init-db-states' into 'main'

Simplify calculation of starting index within indexer

See merge request flarenetwork/FSP/flare-system-c-chain-indexer!74

Author: LukaAvbreht

benchmarks/songbird_test.go

@@ -53,7 +53,7 @@ func BenchmarkBlockRequests(b *testing.B) {
 			logger.Fatal("Indexer create error: %s", err)
 		}
 
-		err = cIndexer.IndexHistory(ctx)
+		_, err = cIndexer.IndexHistory(ctx)
 		if err != nil {
 			logger.Fatal("History run error: %s", err)
 		}

database/states.go

@@ -2,7 +2,6 @@ package database
 
 import (
 	"context"
-	"flare-ftso-indexer/logger"
 	"sync"
 	"time"
 
@@ -64,7 +63,13 @@ func (s *DBStates) updateStates(newStates map[string]*State) {
 
 func (s *DBStates) updateIndex(name string, newIndex, blockTimestamp uint64) {
 	s.mu.Lock()
-	s.States[name].updateIndex(newIndex, blockTimestamp)
+	state := s.States[name]
+	if state == nil {
+		state = &State{Name: name}
+		s.States[name] = state
+	}
+
+	state.updateIndex(newIndex, blockTimestamp)
 	s.mu.Unlock()
 }
 
@@ -76,50 +81,32 @@ func (s *DBStates) updateDB(db *gorm.DB, name string) error {
 }
 
 func (s *DBStates) Update(db *gorm.DB, name string, newIndex, blockTimestamp uint64) error {
-	s.mu.RLock()
-	state := s.States[name]
-	s.mu.RUnlock()
-
-	if state == nil {
-		return errors.Errorf("state %s not found", name)
-	}
-
 	s.updateIndex(name, newIndex, blockTimestamp)
 	return s.updateDB(db, name)
 }
 
 func (s *DBStates) UpdateAtStart(
-	db *gorm.DB, startIndex, startBlockTimestamp, lastChainIndex, lastBlockTimestamp, stopIndex uint64,
-) (uint64, uint64, error) {
-	var err error
-
+	db *gorm.DB, startIndex, startBlockTimestamp, lastChainIndex, lastBlockTimestamp uint64,
+) error {
 	s.mu.RLock()
-	firstIndex := s.States[FirstDatabaseIndexState].Index
-	lastIndex := s.States[LastDatabaseIndexState].Index
+	_, firstDatabaseIndexSet := s.States[FirstDatabaseIndexState]
 	s.mu.RUnlock()
 
-	if startIndex >= firstIndex && startIndex <= lastIndex {
-		logger.Info("Data from blocks %d to %d already in the database", startIndex, lastIndex)
-		startIndex = lastIndex + 1
-	} else {
-		logger.Warn("Data from blocks %d to %d not in the database, starting from %d", startIndex, lastIndex, firstIndex)
-
-		// if startIndex is set before existing data in the DB or a break among saved blocks
-		// in the DB is created, then we change the guaranties about the starting block
-		err = s.Update(db, FirstDatabaseIndexState, startIndex, startBlockTimestamp)
+	// Set the first database index state only if it does not exist yet
+	if !firstDatabaseIndexSet {
+		err := s.Update(db, FirstDatabaseIndexState, startIndex, startBlockTimestamp)
 		if err != nil {
-			return 0, 0, errors.Wrap(err, "states.Update")
+			return errors.Wrap(err, "states.Update(FirstDatabaseIndexState)")
 		}
 	}
 
-	err = s.Update(db, LastChainIndexState, lastChainIndex, lastBlockTimestamp)
+	// Set the state for the current latest chain index
+	err := s.Update(db, LastChainIndexState, lastChainIndex, lastBlockTimestamp)
 	if err != nil {
-		return 0, 0, errors.Wrap(err, "states.Update")
+		return errors.Wrap(err, "states.Update(LastChainIndexState)")
 	}
 
-	lastIndex = min(stopIndex, lastChainIndex)
-
-	return startIndex, lastIndex, nil
+	return nil
 }
 
 func UpdateDBStates(ctx context.Context, db *gorm.DB) (*DBStates, error) {
@@ -148,12 +135,7 @@ func getDBStates(ctx context.Context, db *gorm.DB) (map[string]*State, error) {
 					return errors.Wrap(err, "db.Where")
 				}
 
-				// If the state is not found, create a new one.
-				state = &State{Name: name, Updated: time.Now()}
-				err := db.Create(state).Error
-				if err != nil {
-					return errors.Wrap(err, "db.Create")
-				}
+				return nil
 			}
 
 			mu.Lock()

indexer/indexer.go

@@ -143,35 +143,35 @@ func parseTransactionAddress(address string) common.Address {
 	return common.HexToAddress(address)
 }
 
-func (ci *BlockIndexer) IndexHistory(ctx context.Context) error {
+func (ci *BlockIndexer) IndexHistory(ctx context.Context) (uint64, error) {
 	states, err := database.UpdateDBStates(ctx, ci.db)
 	if err != nil {
-		return errors.Wrap(err, "database.UpdateDBStates")
+		return 0, errors.Wrap(err, "database.UpdateDBStates")
 	}
 
-	ixRange, err := ci.getIndexRange(ctx, states)
+	ixRange, err := ci.getIndexRange(ctx, states, ci.params.StartIndex)
 	if err != nil {
-		return err
+		return 0, err
 	}
 
 	logger.Info("Starting to index blocks from %d to %d", ixRange.start, ixRange.end)
 
 	for i := ixRange.start; i <= ixRange.end; i = i + ci.params.BatchSize {
 		if err := ci.indexBatch(ctx, states, i, ixRange); err != nil {
-			return err
+			return 0, err
 		}
 
 		// in the second to last run of the loop update lastIndex to get the blocks
 		// that were produced during the run of the algorithm
 		if ci.shouldUpdateLastIndex(ixRange, i) {
 			ixRange, err = ci.updateLastIndexHistory(ctx, states, ixRange)
 			if err != nil {
-				return err
+				return 0, err
 			}
 		}
 	}
 
-	return nil
+	return ixRange.end, nil
 }
 
 func (ci *BlockIndexer) indexBatch(
@@ -340,24 +340,27 @@ type indexRange struct {
 }
 
 func (ci *BlockIndexer) getIndexRange(
-	ctx context.Context, states *database.DBStates,
+	ctx context.Context, states *database.DBStates, startIndex uint64,
 ) (*indexRange, error) {
 	lastChainIndex, lastChainTimestamp, err := ci.fetchLastBlockIndex(ctx)
 	if err != nil {
 		return nil, errors.Wrap(err, "ci.fetchLastBlockIndex")
 	}
 
-	startTimestamp, err := ci.fetchBlockTimestamp(ctx, ci.params.StartIndex)
+	startTimestamp, err := ci.fetchBlockTimestamp(ctx, startIndex)
 	if err != nil {
 		return nil, errors.Wrap(err, "ci.fetchBlockTimestamp")
 	}
 
-	startIndex, lastIndex, err := states.UpdateAtStart(ci.db, ci.params.StartIndex,
-		startTimestamp, lastChainIndex, lastChainTimestamp, ci.params.StopIndex)
+	err = states.UpdateAtStart(
+		ci.db, startIndex, startTimestamp, lastChainIndex, lastChainTimestamp,
+	)
 	if err != nil {
 		return nil, errors.Wrap(err, "states.UpdateAtStart")
 	}
 
+	lastIndex := min(lastChainIndex, ci.params.StopIndex)
+
 	return &indexRange{start: startIndex, end: lastIndex}, nil
 }
 
@@ -448,13 +451,13 @@ func (ci *BlockIndexer) updateLastIndexHistory(
 	return ixRange, nil
 }
 
-func (ci *BlockIndexer) IndexContinuous(ctx context.Context) error {
+func (ci *BlockIndexer) IndexContinuous(ctx context.Context, startIndex uint64) error {
 	states, err := database.UpdateDBStates(ctx, ci.db)
 	if err != nil {
 		return errors.Wrap(err, "database.UpdateDBStates")
 	}
 
-	ixRange, err := ci.getIndexRange(ctx, states)
+	ixRange, err := ci.getIndexRange(ctx, states, startIndex)
 	if err != nil {
 		return errors.Wrap(err, "ci.getIndexRange")
 	}
@@ -493,7 +496,7 @@ func (ci *BlockIndexer) IndexContinuous(ctx context.Context) error {
 		blockNum++
 	}
 
-	logger.Debug("Stopping the indexer at block %d", states.States[database.LastDatabaseIndexState].Index)
+	logger.Debug("Stopping the indexer at block %d", blockNum)
 
 	return nil
 }

indexer/indexer_test.go

@@ -165,9 +165,6 @@ func runIndexer(ctx context.Context, mockChain *indexer_testing.MockChain, db *g
 
 	time.Sleep(3 * time.Second)
 
-	// set a new starting index based on the history drop interval
-	historyDropIntervalSeconds := uint64(10000)
-
 	nodeURL, err := cfg.Chain.FullNodeURL()
 	if err != nil {
 		logger.Fatal("Invalid node URL in config: %s", err)
@@ -185,21 +182,16 @@ func runIndexer(ctx context.Context, mockChain *indexer_testing.MockChain, db *g
 	}
 
 	// index history with parallel processing
-	err = cIndexer.IndexHistory(ctx)
+	historyLastIndex, err := cIndexer.IndexHistory(ctx)
 	if err != nil {
 		logger.Fatal("History run error: %s", err)
 	}
 
-	// turn on the function to delete in the database everything that
-	// is older than the historyDrop interval
-	go database.DropHistory(
-		ctx, db, historyDropIntervalSeconds, database.HistoryDropIntervalCheck, ethClient, 0,
-	)
-
-	// run indexer
-	err = cIndexer.IndexContinuous(ctx)
-	if err != nil {
-		logger.Fatal("Continuous run error: %s", err)
+	if historyLastIndex != cfg.Indexer.StopIndex {
+		logger.Fatal(
+			"History indexing did not reach the stop index: last %d, stop %d",
+			historyLastIndex, cfg.Indexer.StopIndex,
+		)
 	}
 
 	return nil

main.go

@@ -155,9 +155,9 @@ func runIndexer(
 		return err
 	}
 
-	err = boff.RetryNoReturn(
+	historyLastIndex, err := boff.Retry(
 		ctx,
-		func() error {
+		func() (uint64, error) {
 			return cIndexer.IndexHistory(ctx)
 		},
 		"IndexHistory",
@@ -180,7 +180,7 @@ func runIndexer(
 	err = boff.RetryNoReturn(
 		ctx,
 		func() error {
-			return cIndexer.IndexContinuous(ctx)
+			return cIndexer.IndexContinuous(ctx, historyLastIndex+1)
 		},
 		"IndexContinuous",
 	)

main_test.go

@@ -74,13 +74,14 @@ func TestIntegrationIndexHistory(t *testing.T) {
 }
 
 func (suite *IntegrationIndexContinuousSuite) SetupSuite() {
-	err := suite.indexer.IndexContinuous(suite.ctx)
+	err := suite.indexer.IndexContinuous(suite.ctx, startBlock)
 	require.NoError(suite.T(), err, "Could not run the indexer")
 }
 
 func (suite *IntegrationIndexHistorySuite) SetupSuite() {
-	err := suite.indexer.IndexHistory(suite.ctx)
+	lastIndex, err := suite.indexer.IndexHistory(suite.ctx)
 	require.NoError(suite.T(), err, "Could not run the indexer")
+	require.Equal(suite.T(), uint64(endBlockHistory), lastIndex, "Last indexed block does not match expected value")
 }
 
 func (suite *IntegrationIndex) TestCheckBlocks() {