Merge branch 'backoff-wait' into 'main'

Timeout/backoff improvements

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

README.md

@@ -57,6 +57,10 @@ console = true
 
 [chain]
 node_url = "http://127.0.0.1:8545/"
+
+[timeout]
+backoff_max_elapsed_time_seconds = 300 # optional, defaults to 300s = 5 minutes. Affects how long the indexer will keep retrying in case of a complete outage of the node provider. Set to 0 to retry indefinitely.
+timeout_milis = 1000  # optional, defaults to 1000ms = 1s. Try increasing if you see timeout errors often.
 ```
 
 ### Database

config.toml

@@ -60,3 +60,7 @@ console = true
 
 [chain]
 node_url = "http://127.0.0.1:8545/"
+
+[timeout]
+backoff_max_elapsed_time_seconds = 300
+timeout_milis = 1000

config/config.go

@@ -13,20 +13,36 @@ import (
 
 var (
 	BackoffMaxElapsedTime time.Duration                = 5 * time.Minute
-	DefaultTimeout        time.Duration                = 1000 * time.Millisecond
+	Timeout               time.Duration                = 1000 * time.Millisecond
 	GlobalConfigCallback  ConfigCallback[GlobalConfig] = ConfigCallback[GlobalConfig]{}
 	CfgFlag                                            = flag.String("config", "config.toml", "Configuration file (toml format)")
 )
 
+func init() {
+	GlobalConfigCallback.AddCallback(func(config GlobalConfig) {
+		tCfg := config.TimeoutConfig()
+
+		if tCfg.BackoffMaxElapsedTimeSeconds != nil {
+			BackoffMaxElapsedTime = time.Duration(*tCfg.BackoffMaxElapsedTimeSeconds) * time.Second
+		}
+
+		if tCfg.TimeoutMillis > 0 {
+			Timeout = time.Duration(tCfg.TimeoutMillis) * time.Millisecond
+		}
+	})
+}
+
 type GlobalConfig interface {
 	LoggerConfig() LoggerConfig
+	TimeoutConfig() TimeoutConfig
 }
 
 type Config struct {
 	DB      DBConfig      `toml:"db"`
 	Logger  LoggerConfig  `toml:"logger"`
 	Chain   ChainConfig   `toml:"chain"`
 	Indexer IndexerConfig `toml:"indexer"`
+	Timeout TimeoutConfig `toml:"timeout"`
 }
 
 type LoggerConfig struct {
@@ -63,6 +79,11 @@ type IndexerConfig struct {
 	CollectLogs         []LogInfo         `toml:"collect_logs"`
 }
 
+type TimeoutConfig struct {
+	BackoffMaxElapsedTimeSeconds *int `toml:"backoff_max_elapsed_time_seconds"`
+	TimeoutMillis                int  `toml:"timeout_millis"`
+}
+
 type TransactionInfo struct {
 	ContractAddress string `toml:"contract_address"`
 	FuncSig         string `toml:"func_sig"`
@@ -106,6 +127,10 @@ func (c Config) LoggerConfig() LoggerConfig {
 	return c.Logger
 }
 
+func (c Config) TimeoutConfig() TimeoutConfig {
+	return c.Timeout
+}
+
 func (cc ChainConfig) FullNodeURL() (*url.URL, error) {
 	u, err := url.Parse(cc.NodeURL)
 	if err != nil {

database/history_drop.go

@@ -114,7 +114,7 @@ func getBlockTimestamp(ctx context.Context, index *big.Int, client *ethclient.Cl
 	var block *types.Block
 	err := backoff.Retry(
 		func() (err error) {
-			ctx, cancelFunc := context.WithTimeout(ctx, config.DefaultTimeout)
+			ctx, cancelFunc := context.WithTimeout(ctx, config.Timeout)
 			defer cancelFunc()
 
 			block, err = client.BlockByNumber(ctx, index)

indexer/blocks.go

@@ -35,15 +35,15 @@ func (ci *BlockIndexer) fetchBlock(ctx context.Context, index *uint64) (block *t
 
 	err = backoff.RetryNotify(
 		func() error {
-			ctx, cancelFunc := context.WithTimeout(ctx, config.DefaultTimeout)
+			ctx, cancelFunc := context.WithTimeout(ctx, config.Timeout)
 			defer cancelFunc()
 
 			block, err = ci.client.BlockByNumber(ctx, indexBigInt)
 			return err
 		},
 		bOff,
 		func(err error, d time.Duration) {
-			logger.Debug("BlockByNumber error: %s after %s", err, d)
+			logger.Debug("BlockByNumber error: %s. Will retry after %s", err, d)
 		},
 	)
 

indexer/logs.go

@@ -75,7 +75,7 @@ func (ci *BlockIndexer) fetchLogsChunk(
 
 	err := backoff.RetryNotify(
 		func() error {
-			ctx, cancelFunc := context.WithTimeout(ctx, config.DefaultTimeout)
+			ctx, cancelFunc := context.WithTimeout(ctx, config.Timeout)
 			defer cancelFunc()
 
 			var err error
@@ -84,7 +84,7 @@ func (ci *BlockIndexer) fetchLogsChunk(
 		},
 		bOff,
 		func(err error, d time.Duration) {
-			logger.Debug("FilterLogs error: %s after %s", err, d)
+			logger.Debug("FilterLogs error: %s. Will retry after %s", err, d)
 		},
 	)
 	if err != nil {

indexer/transactions.go

@@ -71,15 +71,15 @@ func (ci *BlockIndexer) getTransactionsReceipt(
 		if policy.status || policy.collectEvents {
 			err := backoff.RetryNotify(
 				func() (err error) {
-					ctx, cancelFunc := context.WithTimeout(ctx, config.DefaultTimeout)
+					ctx, cancelFunc := context.WithTimeout(ctx, config.Timeout)
 					defer cancelFunc()
 
 					receipt, err = ci.client.TransactionReceipt(ctx, tx.Hash())
 					return err
 				},
 				bOff,
 				func(err error, d time.Duration) {
-					logger.Error("TransactionReceipt error: %s after %s", err, d)
+					logger.Error("TransactionReceipt error: %s. Will retry after %s", err, d)
 				},
 			)
 

main.go

@@ -80,7 +80,7 @@ func runIndexer(ctx context.Context, cfg *config.Config, db *gorm.DB, ethClient
 		},
 		bOff,
 		func(err error, d time.Duration) {
-			logger.Error("Index history error: %s after %s", err, d)
+			logger.Error("Index history error: %s. Will retry after %s", err, d)
 		},
 	)
 	if err != nil {
@@ -99,7 +99,7 @@ func runIndexer(ctx context.Context, cfg *config.Config, db *gorm.DB, ethClient
 		},
 		bOff,
 		func(err error, d time.Duration) {
-			logger.Error("Index continuous error: %s after %s", err, d)
+			logger.Error("Index continuous error: %s. Will retry after %s", err, d)
 		},
 	)
 	if err != nil {