chain validation and fix command

Author: iuwqyir

cmd/root.go

@@ -198,6 +198,7 @@ func init() {
 	viper.BindPFlag("publisher.events.topic0Filter", rootCmd.PersistentFlags().Lookup("publisher-events-topic0Filter"))
 	rootCmd.AddCommand(orchestratorCmd)
 	rootCmd.AddCommand(apiCmd)
+	rootCmd.AddCommand(validateCmd)
 }
 
 func initConfig() {

cmd/validate.go

@@ -0,0 +1,134 @@
+package cmd
+
+import (
+	"crypto/tls"
+	"fmt"
+	"math/big"
+	"strconv"
+
+	"github.com/ClickHouse/clickhouse-go/v2"
+	"github.com/rs/zerolog/log"
+	"github.com/spf13/cobra"
+	config "github.com/thirdweb-dev/indexer/configs"
+	"github.com/thirdweb-dev/indexer/internal/rpc"
+	"github.com/thirdweb-dev/indexer/internal/storage"
+	"github.com/thirdweb-dev/indexer/internal/validation"
+)
+
+var (
+	validateCmd = &cobra.Command{
+		Use:   "validate",
+		Short: "TBD",
+		Long:  "TBD",
+		Run: func(cmd *cobra.Command, args []string) {
+			RunValidate(cmd, args)
+		},
+	}
+)
+
+func RunValidate(cmd *cobra.Command, args []string) {
+	batchSize := big.NewInt(1000)
+	fixBatchSize := 0 // default is no batch size
+	if len(args) > 0 {
+		batchSizeFromArgs, err := strconv.Atoi(args[0])
+		if err != nil {
+			log.Fatal().Err(err).Msg("Failed to parse batch size")
+		}
+		if batchSizeFromArgs < 1 {
+			batchSizeFromArgs = 1
+		}
+		batchSize = big.NewInt(int64(batchSizeFromArgs))
+		log.Info().Msgf("Using batch size %d from args", batchSize)
+	}
+	if len(args) > 1 {
+		fixBatchSizeFromArgs, err := strconv.Atoi(args[1])
+		if err != nil {
+			log.Fatal().Err(err).Msg("Failed to parse fix batch size")
+		}
+		fixBatchSize = fixBatchSizeFromArgs
+	}
+	log.Debug().Msgf("Batch size: %d, fix batch size: %d", batchSize, fixBatchSize)
+	batchSize = new(big.Int).Sub(batchSize, big.NewInt(1)) // -1 because range ends are inclusive
+
+	rpcClient, err := rpc.Initialize()
+	if err != nil {
+		log.Fatal().Err(err).Msg("Failed to initialize RPC")
+	}
+	log.Info().Msgf("Running validation for chain %d", rpcClient.GetChainID())
+
+	s, err := storage.NewStorageConnector(&config.Cfg.Storage)
+	if err != nil {
+		log.Fatal().Err(err).Msg("Failed to initialize storage")
+	}
+	cursor, err := validation.InitCursor(rpcClient.GetChainID(), s)
+	if err != nil {
+		log.Fatal().Err(err).Msg("Failed to initialize cursor")
+	}
+	log.Debug().Msgf("Cursor initialized for chain %d, starting from block %d", rpcClient.GetChainID(), cursor.LastScannedBlockNumber)
+
+	conn, err := clickhouse.Open(&clickhouse.Options{
+		Addr:     []string{fmt.Sprintf("%s:%d", config.Cfg.Storage.Main.Clickhouse.Host, config.Cfg.Storage.Main.Clickhouse.Port)},
+		Protocol: clickhouse.Native,
+		TLS:      &tls.Config{},
+		Auth: clickhouse.Auth{
+			Username: config.Cfg.Storage.Main.Clickhouse.Username,
+			Password: config.Cfg.Storage.Main.Clickhouse.Password,
+		},
+		Settings: func() clickhouse.Settings {
+			settings := clickhouse.Settings{
+				"do_not_merge_across_partitions_select_final": "1",
+				"use_skip_indexes_if_final":                   "1",
+				"optimize_move_to_prewhere_if_final":          "1",
+				"async_insert":                                "1",
+				"wait_for_async_insert":                       "1",
+			}
+			return settings
+		}(),
+	})
+	if err != nil {
+		log.Fatal().Err(err).Msg("Failed to connect to ClickHouse")
+	}
+	defer conn.Close()
+
+	startBlock := new(big.Int).Add(cursor.LastScannedBlockNumber, big.NewInt(1))
+
+	for startBlock.Cmp(cursor.MaxBlockNumber) <= 0 {
+		batchEndBlock := new(big.Int).Add(startBlock, batchSize)
+		if batchEndBlock.Cmp(cursor.MaxBlockNumber) > 0 {
+			batchEndBlock = new(big.Int).Set(cursor.MaxBlockNumber)
+		}
+
+		log.Info().Msgf("Validating batch of blocks from %s to %s", startBlock.String(), batchEndBlock.String())
+		err := validateAndFixRange(rpcClient, s, conn, startBlock, batchEndBlock, fixBatchSize)
+		if err != nil {
+			log.Fatal().Err(err).Msgf("failed to validate range %v-%v", startBlock, batchEndBlock)
+		}
+
+		startBlock = new(big.Int).Add(batchEndBlock, big.NewInt(1))
+		cursor.Update(batchEndBlock)
+	}
+}
+
+/**
+ * Validates a range of blocks (end and start are inclusive) for a given chain and fixes any problems it finds
+ */
+func validateAndFixRange(rpcClient rpc.IRPCClient, s storage.IStorage, conn clickhouse.Conn, startBlock *big.Int, endBlock *big.Int, fixBatchSize int) error {
+	chainId := rpcClient.GetChainID()
+	err := validation.FindAndRemoveDuplicates(conn, chainId, startBlock, endBlock)
+	if err != nil {
+		log.Fatal().Err(err).Msg("Failed to find and fix duplicates")
+	}
+
+	err = validation.FindAndFixGaps(rpcClient, s, conn, chainId, startBlock, endBlock)
+	if err != nil {
+		log.Fatal().Err(err).Msg("Failed to find and fix gaps")
+	}
+
+	err = validation.ValidateAndFixBlocks(rpcClient, s, conn, startBlock, endBlock, fixBatchSize)
+	if err != nil {
+		log.Fatal().Err(err).Msg("Failed to validate and fix blocks")
+	}
+
+	log.Debug().Msgf("Validation complete for range %v-%v", startBlock, endBlock)
+	return nil
+}

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/ethereum/go-ethereum v1.14.8
 	github.com/gin-gonic/gin v1.10.0
 	github.com/gorilla/schema v1.4.1
+	github.com/holiman/uint256 v1.3.1
 	github.com/prometheus/client_golang v1.20.4
 	github.com/rs/zerolog v1.33.0
 	github.com/spf13/cobra v1.8.1
@@ -20,6 +21,7 @@ require (
 
 require (
 	github.com/ClickHouse/ch-go v0.63.1 // indirect
+	github.com/DataDog/zstd v1.4.5 // indirect
 	github.com/KyleBanks/depth v1.2.1 // indirect
 	github.com/Microsoft/go-winio v0.6.2 // indirect
 	github.com/andybalholm/brotli v1.1.1 // indirect
@@ -31,16 +33,25 @@ require (
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudwego/base64x v0.1.4 // indirect
 	github.com/cloudwego/iasm v0.2.0 // indirect
+	github.com/cockroachdb/errors v1.11.3 // indirect
+	github.com/cockroachdb/fifo v0.0.0-20240606204812-0bbfbd93a7ce // indirect
+	github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b // indirect
+	github.com/cockroachdb/pebble v1.1.1 // indirect
+	github.com/cockroachdb/redact v1.1.5 // indirect
+	github.com/cockroachdb/tokenbucket v0.0.0-20230807174530-cc333fc44b06 // indirect
 	github.com/consensys/bavard v0.1.13 // indirect
 	github.com/consensys/gnark-crypto v0.12.1 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.5 // indirect
+	github.com/crate-crypto/go-ipa v0.0.0-20240223125850-b1e8a79f509c // indirect
 	github.com/crate-crypto/go-kzg-4844 v1.0.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/deckarep/golang-set/v2 v2.6.0 // indirect
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1 // indirect
 	github.com/ethereum/c-kzg-4844 v1.0.0 // indirect
+	github.com/ethereum/go-verkle v0.1.1-0.20240306133620-7d920df305f0 // indirect
 	github.com/fsnotify/fsnotify v1.7.0 // indirect
 	github.com/gabriel-vasile/mimetype v1.4.7 // indirect
+	github.com/getsentry/sentry-go v0.27.0 // indirect
 	github.com/gin-contrib/sse v0.1.0 // indirect
 	github.com/go-faster/city v1.0.1 // indirect
 	github.com/go-faster/errors v0.7.1 // indirect
@@ -53,26 +64,32 @@ require (
 	github.com/go-playground/universal-translator v0.18.1 // indirect
 	github.com/go-playground/validator/v10 v10.23.0 // indirect
 	github.com/goccy/go-json v0.10.4 // indirect
+	github.com/gofrs/flock v0.8.1 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang-jwt/jwt/v4 v4.5.1 // indirect
+	github.com/golang/snappy v0.0.5-0.20220116011046-fa5810519dcb // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/gorilla/websocket v1.4.2 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
-	github.com/holiman/uint256 v1.3.1 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/compress v1.17.11 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.9 // indirect
+	github.com/kr/pretty v0.3.1 // indirect
+	github.com/kr/text v0.2.0 // indirect
 	github.com/leodido/go-urn v1.4.0 // indirect
 	github.com/magiconair/properties v1.8.7 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/mattn/go-runewidth v0.0.13 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/mmcloughlin/addchain v0.4.0 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/olekukonko/tablewriter v0.0.5 // indirect
 	github.com/paulmach/orb v0.11.1 // indirect
 	github.com/pelletier/go-toml/v2 v2.2.3 // indirect
 	github.com/pierrec/lz4/v4 v4.1.22 // indirect
@@ -81,6 +98,8 @@ require (
 	github.com/prometheus/client_model v0.6.1 // indirect
 	github.com/prometheus/common v0.55.0 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect
+	github.com/rivo/uniseg v0.2.0 // indirect
+	github.com/rogpeppe/go-internal v1.11.0 // indirect
 	github.com/sagikazarmark/locafero v0.4.0 // indirect
 	github.com/sagikazarmark/slog-shim v0.1.0 // indirect
 	github.com/segmentio/asm v1.2.0 // indirect
@@ -93,6 +112,7 @@ require (
 	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/subosito/gotenv v1.6.0 // indirect
 	github.com/supranational/blst v0.3.11 // indirect
+	github.com/syndtr/goleveldb v1.0.1-0.20210819022825-2ae1ddf74ef7 // indirect
 	github.com/tklauser/go-sysconf v0.3.12 // indirect
 	github.com/tklauser/numcpus v0.6.1 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect