lag by blocks for live polling

Author: nischitpra

configs/config.go

@@ -61,6 +61,7 @@ type Config struct {
 	CommitterCompressionThresholdMB  int       `env:"COMMITTER_COMPRESSION_THRESHOLD_MB" envDefault:"50"`
 	CommitterKafkaBatchSize          int       `env:"COMMITTER_KAFKA_BATCH_SIZE" envDefault:"500"`
 	CommitterIsLive                  bool      `env:"COMMITTER_IS_LIVE" envDefault:"false"`
+	CommitterLagByBlocks             uint64    `env:"COMMITTER_LAG_BY_BLOCKS" envDefault:"0"`
 	StagingS3Bucket                  string    `env:"STAGING_S3_BUCKET" envDefault:"thirdweb-insight-production"`
 	StagingS3Region                  string    `env:"STAGING_S3_REGION" envDefault:"us-west-2"`
 	StagingS3AccessKeyID             string    `env:"STAGING_S3_ACCESS_KEY_ID"`

internal/committer/poollatest.go

@@ -29,7 +29,7 @@ func pollLatest() error {
 		// Update latest block number metric
 		metrics.CommitterLatestBlockNumber.WithLabelValues(indexerName, chainIdStr).Set(float64(latestBlock.Uint64()))
 
-		if nextBlockNumber >= latestBlock.Uint64() {
+		if nextBlockNumber+config.Cfg.CommitterLagByBlocks >= latestBlock.Uint64() {
 			time.Sleep(250 * time.Millisecond)
 			continue
 		}