committer start block

Author: nischitpra

cmd/committer.go

@@ -7,19 +7,31 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/rs/zerolog/log"
 	"github.com/spf13/cobra"
+	config "github.com/thirdweb-dev/indexer/configs"
 	"github.com/thirdweb-dev/indexer/internal/committer"
 )
 
+var committerStartBlock uint64
+
 var committerCmd = &cobra.Command{
 	Use:   "committer",
 	Short: "run committer",
 	Long:  "published data from s3 to kafka. if block is not found in s3, it will panic",
 	Run:   RunCommitter,
 }
 
+func init() {
+	committerCmd.Flags().Uint64Var(&committerStartBlock, "start-block", 0, "start committing from this block number (overrides ClickHouse max block when > 0)")
+}
+
 func RunCommitter(cmd *cobra.Command, args []string) {
 	fmt.Println("running committer")
 
+	if committerStartBlock > 0 {
+		config.Cfg.CommitterStartBlock = committerStartBlock
+		log.Info().Uint64("start_block", committerStartBlock).Msg("Committer start block override enabled")
+	}
+
 	// Start Prometheus metrics server
 	log.Info().Msg("Starting Metrics Server on port 2112")
 	go func() {

configs/config.go

@@ -62,6 +62,10 @@ type Config struct {
 	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"`
+	// CommitterStartBlock, when set (>0), forces the committer to start publishing
+	// from this block number regardless of what ClickHouse says is already committed.
+	// This can cause duplicate publishing if ClickHouse already contains higher blocks.
+	CommitterStartBlock              uint64    `env:"COMMITTER_START_BLOCK" 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/committer.go

@@ -104,6 +104,20 @@ func getLastTrackedBlockNumberAndBlockRangesFromS3() (int64, []types.BlockRange,
 	}
 	log.Debug().Int64("max_block_number", maxBlockNumber).Msg("Retrieved max block number from ClickHouse.(-1 means nothing committed yet, start from 0)")
 
+	// Optional override: force the committer to start from a specific block number.
+	// We implement this by pretending ClickHouse max is (startBlock - 1), so both S3
+	// range scanning and live RPC polling begin at startBlock.
+	if config.Cfg.CommitterStartBlock > 0 {
+		overrideMax := int64(config.Cfg.CommitterStartBlock) - 1
+		if maxBlockNumber < overrideMax {
+			maxBlockNumber = overrideMax
+			log.Info().
+				Int64("clickhouse_max_block", maxBlockNumber).
+				Uint64("override_start_block", config.Cfg.CommitterStartBlock).
+				Msg("CommitterStartBlock override enabled; starting earlier than ClickHouse cursor")
+		}
+	}
+
 	blockRanges, err := libs.GetBlockRangesFromS3(maxBlockNumber)
 	if err != nil {
 		log.Error().Err(err).Msg("Failed to get block ranges from S3")