Merge branch 'block-confirmations' into 'main'

Set configurable number of block confirmations

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

Author: ryanc-flare

config/config.go

@@ -18,6 +18,8 @@ var (
 	CfgFlag                                            = flag.String("config", "config.toml", "Configuration file (toml format)")
 )
 
+const defaultConfirmations = 1
+
 func init() {
 	GlobalConfigCallback.AddCallback(func(config GlobalConfig) {
 		tCfg := config.TimeoutConfig()
@@ -77,6 +79,7 @@ type IndexerConfig struct {
 	NewBlockCheckMillis int               `toml:"new_block_check_millis"`
 	CollectTransactions []TransactionInfo `toml:"collect_transactions"`
 	CollectLogs         []LogInfo         `toml:"collect_logs"`
+	Confirmations       uint64            `toml:"confirmations"`
 }
 
 type TimeoutConfig struct {
@@ -99,7 +102,7 @@ type LogInfo struct {
 func BuildConfig() (*Config, error) {
 	cfgFileName := *CfgFlag
 
-	cfg := new(Config)
+	cfg := &Config{Indexer: IndexerConfig{Confirmations: defaultConfirmations}}
 	err := parseConfigFile(cfg, cfgFileName)
 	if err != nil {
 		return nil, err

indexer/.snapshots/TestIndexer

@@ -0,0 +1,49 @@
+(*database.DBStates)({
+  States: (map[string]*database.State) (len=3) {
+    (string) (len=20) "first_database_block": (*database.State)({
+      BaseEntity: (database.BaseEntity) {
+        ID: (uint64) 1
+      },
+      Name: (string) (len=20) "first_database_block",
+      Index: (uint64) 1112,
+      BlockTimestamp: (uint64) 1662557489,
+      Updated: (time.Time) 0001-01-01 00:00:00 +0000 UTC
+    }),
+    (string) (len=16) "last_chain_block": (*database.State)({
+      BaseEntity: (database.BaseEntity) {
+        ID: (uint64) 3
+      },
+      Name: (string) (len=16) "last_chain_block",
+      Index: (uint64) 10223878,
+      BlockTimestamp: (uint64) 1721830576,
+      Updated: (time.Time) 0001-01-01 00:00:00 +0000 UTC
+    }),
+    (string) (len=19) "last_database_block": (*database.State)({
+      BaseEntity: (database.BaseEntity) {
+        ID: (uint64) 2
+      },
+      Name: (string) (len=19) "last_database_block",
+      Index: (uint64) 2400,
+      BlockTimestamp: (uint64) 1662560050,
+      Updated: (time.Time) 0001-01-01 00:00:00 +0000 UTC
+    })
+  },
+  mu: (sync.RWMutex) {
+    w: (sync.Mutex) {
+      state: (int32) 0,
+      sema: (uint32) 0
+    },
+    writerSem: (uint32) 0,
+    readerSem: (uint32) 0,
+    readerCount: (atomic.Int32) {
+      _: (atomic.noCopy) {
+      },
+      v: (int32) 0
+    },
+    readerWait: (atomic.Int32) {
+      _: (atomic.noCopy) {
+      },
+      v: (int32) 0
+    }
+  }
+})

indexer/blocks.go

@@ -5,7 +5,6 @@ import (
 	"flare-ftso-indexer/config"
 	"flare-ftso-indexer/database"
 	"flare-ftso-indexer/logger"
-	"fmt"
 	"math/big"
 	"sync"
 	"time"
@@ -26,10 +25,7 @@ func newBlockBatch(batchSize uint64) *blockBatch {
 }
 
 func (ci *BlockIndexer) fetchBlock(ctx context.Context, index *uint64) (block *types.Block, err error) {
-	var indexBigInt *big.Int
-	if index != nil {
-		indexBigInt = new(big.Int).SetUint64(*index)
-	}
+	indexBigInt := indexToBigInt(index)
 
 	bOff := backoff.NewExponentialBackOff()
 	bOff.MaxElapsedTime = config.BackoffMaxElapsedTime
@@ -55,22 +51,63 @@ func (ci *BlockIndexer) fetchBlock(ctx context.Context, index *uint64) (block *t
 	return block, nil
 }
 
+func (ci *BlockIndexer) fetchBlockHeader(ctx context.Context, index *uint64) (header *types.Header, err error) {
+	indexBigInt := indexToBigInt(index)
+
+	bOff := backoff.NewExponentialBackOff()
+	bOff.MaxElapsedTime = config.BackoffMaxElapsedTime
+
+	err = backoff.RetryNotify(
+		func() error {
+			ctx, cancelFunc := context.WithTimeout(ctx, config.Timeout)
+			defer cancelFunc()
+
+			header, err = ci.client.HeaderByNumber(ctx, indexBigInt)
+			return err
+		},
+		bOff,
+		func(err error, d time.Duration) {
+			logger.Debug("HeaderByNumber error: %s. Will retry after %s", err, d)
+		},
+	)
+
+	if err != nil {
+		return nil, errors.Wrap(err, "ci.client.HeaderByNumber")
+	}
+
+	return header, nil
+}
+
+func indexToBigInt(index *uint64) *big.Int {
+	if index == nil {
+		return nil
+	}
+
+	return new(big.Int).SetUint64(*index)
+}
+
 func (ci *BlockIndexer) fetchLastBlockIndex(ctx context.Context) (uint64, uint64, error) {
-	lastBlock, err := ci.fetchBlock(ctx, nil)
+	lastBlock, err := ci.fetchBlockHeader(ctx, nil)
+	if err != nil {
+		return 0, 0, errors.Wrap(err, "fetchBlockHeader last")
+	}
+
+	latestConfirmedNumber := lastBlock.Number.Uint64() - ci.params.Confirmations
+	latestConfirmedHeader, err := ci.fetchBlockHeader(ctx, &latestConfirmedNumber)
 	if err != nil {
-		return 0, 0, fmt.Errorf("fetchLastBlockIndex: %w", err)
+		return 0, 0, errors.Wrap(err, "fetchBlockHeader latestConfirmed")
 	}
 
-	return lastBlock.NumberU64(), lastBlock.Time(), nil
+	return latestConfirmedNumber, latestConfirmedHeader.Time, nil
 }
 
 func (ci *BlockIndexer) fetchBlockTimestamp(ctx context.Context, index uint64) (uint64, error) {
-	lastBlock, err := ci.fetchBlock(ctx, &index)
+	lastBlock, err := ci.fetchBlockHeader(ctx, &index)
 	if err != nil {
-		return 0, fmt.Errorf("fetchBlockTimestamp: %w", err)
+		return 0, errors.Wrap(err, "fetchBlockHeader")
 	}
 
-	return lastBlock.Time(), nil
+	return lastBlock.Time, nil
 }
 
 func (ci *BlockIndexer) processBlocks(

indexer/indexer_test.go

@@ -11,17 +11,21 @@ import (
 	"time"
 
 	"github.com/ava-labs/coreth/ethclient"
+	"github.com/bradleyjkemp/cupaloy/v2"
 	"github.com/caarlos0/env/v10"
-	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gorm.io/gorm"
 )
 
 type testConfig struct {
-	DBHost        string `env:"DB_HOST" envDefault:"localhost"`
-	DBPort        int    `env:"DB_PORT" envDefault:"3306"`
-	DBName        string `env:"DB_NAME" envDefault:"flare_ftso_indexer_test"`
-	DBUsername    string `env:"DB_USERNAME" envDefault:"root"`
-	DBPassword    string `env:"DB_PASSWORD" envDefault:"root"`
-	MockChainPort int    `env:"MOCK_CHAIN_PORT" envDefault:"5500"`
+	DBHost          string `env:"DB_HOST" envDefault:"localhost"`
+	DBPort          int    `env:"DB_PORT" envDefault:"3306"`
+	DBName          string `env:"DB_NAME" envDefault:"flare_ftso_indexer_test"`
+	DBUsername      string `env:"DB_USERNAME" envDefault:"root"`
+	DBPassword      string `env:"DB_PASSWORD" envDefault:"root"`
+	MockChainPort   int    `env:"MOCK_CHAIN_PORT" envDefault:"5500"`
+	RecorderNodeURL string `env:"RECORDER_NODE_URL"`
+	ResponsesFile   string `env:"RESPONSES_FILE" envDefault:"../testing/chain_copy/responses.json"`
 }
 
 func TestIndexer(t *testing.T) {
@@ -33,21 +37,6 @@ func TestIndexer(t *testing.T) {
 		t.Fatal("Config parse error:", err)
 	}
 
-	// mock blockchain
-	go func() {
-		err := indexer_testing.MockChain(
-			tCfg.MockChainPort,
-			"../testing/chain_copy/blocks.json",
-			"../testing/chain_copy/transactions.json",
-		)
-		if err != nil {
-			logger.Fatal("Mock chain error: %s", err)
-		}
-	}()
-
-	time.Sleep(3 * time.Second)
-	indexer_testing.ChainLastBlock = 2000
-
 	// set configuration parameters
 	mockChainAddress := fmt.Sprintf("http://localhost:%d", tCfg.MockChainPort)
 	cfgChain := config.ChainConfig{NodeURL: mockChainAddress}
@@ -98,12 +87,51 @@ func TestIndexer(t *testing.T) {
 	cfg := config.Config{Indexer: cfgIndexer, Chain: cfgChain, Logger: cfgLog, DB: cfgDB}
 	config.GlobalConfigCallback.Call(cfg)
 
+	// mock blockchain
+	mockChain, err := indexer_testing.NewMockChain(
+		tCfg.MockChainPort,
+		tCfg.ResponsesFile,
+		tCfg.RecorderNodeURL,
+	)
+	require.NoError(t, err)
+
 	// connect to the database
-	db, err := database.ConnectAndInitialize(ctx, &cfgDB)
+	db, err := database.ConnectAndInitialize(ctx, &cfg.DB)
 	if err != nil {
 		logger.Fatal("Database connect and initialize error: %s", err)
 	}
 
+	err = runIndexer(ctx, mockChain, db, &cfg)
+	require.NoError(t, err)
+
+	// correctness check
+	states, err := database.UpdateDBStates(ctx, db)
+	require.NoError(t, err)
+
+	// Set the update timestamps to zero for the snapshot as these will
+	// vary with current system  time.
+	for _, state := range states.States {
+		state.Updated = time.Time{}
+	}
+
+	cupaloy.SnapshotT(t, states)
+}
+
+func runIndexer(ctx context.Context, mockChain *indexer_testing.MockChain, db *gorm.DB, cfg *config.Config) error {
+	go func() {
+		if err := mockChain.Run(ctx); err != nil {
+			logger.Fatal("Mock chain error: %s", err)
+		}
+	}()
+
+	defer func() {
+		if err := mockChain.Stop(); err != nil {
+			logger.Error("Mock chain stop error: %s", err)
+		}
+	}()
+
+	time.Sleep(3 * time.Second)
+
 	// set a new starting index based on the history drop interval
 	historyDropIntervalSeconds := uint64(10000)
 
@@ -113,7 +141,7 @@ func TestIndexer(t *testing.T) {
 	}
 
 	// create the indexer
-	cIndexer, err := CreateBlockIndexer(&cfg, db, ethClient)
+	cIndexer, err := CreateBlockIndexer(cfg, db, ethClient)
 	if err != nil {
 		logger.Fatal("Create indexer error: %s", err)
 	}
@@ -124,9 +152,6 @@ func TestIndexer(t *testing.T) {
 		logger.Fatal("History run error: %s", err)
 	}
 
-	// at the mock server add new blocks after some time
-	go increaseLastBlockAndStop()
-
 	// turn on the function to delete in the database everything that
 	// is older than the historyDrop interval
 	go database.DropHistory(
@@ -139,19 +164,5 @@ func TestIndexer(t *testing.T) {
 		logger.Fatal("Continuous run error: %s", err)
 	}
 
-	// correctness check
-	states, err := database.UpdateDBStates(ctx, db)
-	assert.NoError(t, err)
-	assert.Equal(t, 1112, int(states.States[database.FirstDatabaseIndexState].Index))
-	assert.Equal(t, 2400, int(states.States[database.LastDatabaseIndexState].Index))
-	assert.Equal(t, 8980059, int(states.States[database.LastChainIndexState].Index))
-}
-
-func increaseLastBlockAndStop() {
-	indexer_testing.ChainLastBlock = 2100
-	time.Sleep(time.Second)
-	indexer_testing.ChainLastBlock = 2200
-	time.Sleep(time.Second)
-	indexer_testing.ChainLastBlock = 2499
-	time.Sleep(10 * time.Second)
+	return nil
 }