Event tracker starting block support (#1290)

* Implement support for specifiying event tracker starting block per contract

* Rename flag

command/genesis/genesis.go

@@ -190,6 +190,14 @@ func setFlags(cmd *cobra.Command) {
 			defaultEpochReward,
 			"reward size for block sealing",
 		)
+
+		cmd.Flags().StringArrayVar(
+			&params.eventTrackerStartBlocks,
+			trackerStartBlocksFlag,
+			[]string{},
+			"event tracker starting block configuration, which is specified per contract address "+
+				"(format: <contract address>:<start block>)",
+		)
 	}
 }
 

command/genesis/params.go

@@ -77,12 +77,13 @@ type genesisParams struct {
 	genesisConfig *chain.Chain
 
 	// PolyBFT
-	manifestPath      string
-	validatorSetSize  int
-	sprintSize        uint64
-	blockTime         time.Duration
-	bridgeJSONRPCAddr string
-	epochReward       uint64
+	manifestPath            string
+	validatorSetSize        int
+	sprintSize              uint64
+	blockTime               time.Duration
+	bridgeJSONRPCAddr       string
+	epochReward             uint64
+	eventTrackerStartBlocks []string
 }
 
 func (p *genesisParams) validateFlags() error {

command/genesis/polybft_params.go

@@ -23,11 +23,12 @@ import (
 )
 
 const (
-	manifestPathFlag     = "manifest"
-	validatorSetSizeFlag = "validator-set-size"
-	sprintSizeFlag       = "sprint-size"
-	blockTimeFlag        = "block-time"
-	bridgeFlag           = "bridge-json-rpc"
+	manifestPathFlag       = "manifest"
+	validatorSetSizeFlag   = "validator-set-size"
+	sprintSizeFlag         = "sprint-size"
+	blockTimeFlag          = "block-time"
+	bridgeFlag             = "bridge-json-rpc"
+	trackerStartBlocksFlag = "tracker-start-blocks"
 
 	defaultManifestPath     = "./manifest.json"
 	defaultEpochSize        = uint64(10)
@@ -56,12 +57,18 @@ func (p *genesisParams) generatePolyBftChainConfig() error {
 		return errNoGenesisValidators
 	}
 
+	eventTrackerStartBlock, err := parseTrackerStartBlocks(params.eventTrackerStartBlocks)
+	if err != nil {
+		return err
+	}
+
 	var bridge *polybft.BridgeConfig
 
 	// populate bridge configuration
 	if p.bridgeJSONRPCAddr != "" && manifest.RootchainConfig != nil {
 		bridge = manifest.RootchainConfig.ToBridgeConfig()
 		bridge.JSONRPCEndpoint = p.bridgeJSONRPCAddr
+		bridge.EventTrackerStartBlocks = eventTrackerStartBlock
 	}
 
 	polyBftConfig := &polybft.PolyBFTConfig{

command/genesis/utils.go

@@ -42,11 +42,6 @@ func (g *GenesisGenError) GetType() string {
 	return g.errorType
 }
 
-type premineInfo struct {
-	address types.Address
-	balance *big.Int
-}
-
 // verifyGenesisExistence checks if the genesis file at the specified path is present
 func verifyGenesisExistence(genesisPath string) *GenesisGenError {
 	_, err := os.Stat(genesisPath)
@@ -67,6 +62,11 @@ func verifyGenesisExistence(genesisPath string) *GenesisGenError {
 	return nil
 }
 
+type premineInfo struct {
+	address types.Address
+	balance *big.Int
+}
+
 // parsePremineInfo parses provided premine information and returns premine address and premine balance
 func parsePremineInfo(premineInfoRaw string) (*premineInfo, error) {
 	address := types.ZeroAddress
@@ -88,6 +88,33 @@ func parsePremineInfo(premineInfoRaw string) (*premineInfo, error) {
 	return &premineInfo{address: address, balance: amount}, nil
 }
 
+// parseTrackerStartBlocks parses provided event tracker start blocks configuration.
+// It is set in a following format: <contractAddress>:<startBlock>.
+// In case smart contract address isn't provided in the string, it is assumed its starting block is 0 implicitly.
+func parseTrackerStartBlocks(trackerStartBlocksRaw []string) (map[types.Address]uint64, error) {
+	trackerStartBlocksConfig := make(map[types.Address]uint64, len(trackerStartBlocksRaw))
+
+	for _, startBlockRaw := range trackerStartBlocksRaw {
+		delimiterIdx := strings.Index(startBlockRaw, ":")
+		if delimiterIdx == -1 {
+			return nil, fmt.Errorf("invalid event tracker start block configuration provided: %s", trackerStartBlocksRaw)
+		}
+
+		// <contractAddress>:<startBlock>
+		address := types.StringToAddress(startBlockRaw[:delimiterIdx])
+		startBlockRaw := startBlockRaw[delimiterIdx+1:]
+
+		startBlock, err := strconv.ParseUint(startBlockRaw, 10, 64)
+		if err != nil {
+			return nil, fmt.Errorf("failed to parse provided start block %s: %w", startBlockRaw, err)
+		}
+
+		trackerStartBlocksConfig[address] = startBlock
+	}
+
+	return trackerStartBlocksConfig, nil
+}
+
 // GetValidatorKeyFiles returns file names which has validator secrets
 func GetValidatorKeyFiles(rootDir, filePrefix string) ([]string, error) {
 	if rootDir == "" {

consensus/polybft/consensus_runtime.go

@@ -157,12 +157,14 @@ func (c *consensusRuntime) close() {
 // if bridge is not enabled, then a dummy state sync manager will be used
 func (c *consensusRuntime) initStateSyncManager(logger hcf.Logger) error {
 	if c.IsBridgeEnabled() {
+		stateSenderAddr := c.config.PolyBFTConfig.Bridge.BridgeAddr
 		stateSyncManager, err := NewStateSyncManager(
 			logger,
 			c.config.State,
 			&stateSyncConfig{
 				key:                   c.config.Key,
-				stateSenderAddr:       c.config.PolyBFTConfig.Bridge.BridgeAddr,
+				stateSenderAddr:       stateSenderAddr,
+				stateSenderStartBlock: c.config.PolyBFTConfig.Bridge.EventTrackerStartBlocks[stateSenderAddr],
 				jsonrpcAddr:           c.config.PolyBFTConfig.Bridge.JSONRPCEndpoint,
 				dataDir:               c.config.DataDir,
 				topic:                 c.config.bridgeTopic,

consensus/polybft/polybft_config.go

@@ -64,11 +64,12 @@ func GetPolyBFTConfig(chainConfig *chain.Chain) (PolyBFTConfig, error) {
 
 // BridgeConfig is the rootchain bridge configuration
 type BridgeConfig struct {
-	BridgeAddr             types.Address `json:"stateSenderAddr"`
-	CheckpointAddr         types.Address `json:"checkpointAddr"`
-	RootERC20PredicateAddr types.Address `json:"rootERC20PredicateAddr"`
-	RootNativeERC20Addr    types.Address `json:"rootNativeERC20Addr"`
-	JSONRPCEndpoint        string        `json:"jsonRPCEndpoint"`
+	BridgeAddr              types.Address            `json:"stateSenderAddr"`
+	CheckpointAddr          types.Address            `json:"checkpointAddr"`
+	RootERC20PredicateAddr  types.Address            `json:"rootERC20PredicateAddr"`
+	RootNativeERC20Addr     types.Address            `json:"rootNativeERC20Addr"`
+	JSONRPCEndpoint         string                   `json:"jsonRPCEndpoint"`
+	EventTrackerStartBlocks map[types.Address]uint64 `json:"eventTrackerStartBlocks"`
 }
 
 func (p *PolyBFTConfig) IsBridgeEnabled() bool {

consensus/polybft/state_sync_manager.go

@@ -60,6 +60,7 @@ func (n *dummyStateSyncManager) GetStateSyncProof(stateSyncID uint64) (types.Pro
 // stateSyncConfig holds the configuration data of state sync manager
 type stateSyncConfig struct {
 	stateSenderAddr       types.Address
+	stateSenderStartBlock uint64
 	jsonrpcAddr           string
 	dataDir               string
 	topic                 topic
@@ -132,6 +133,7 @@ func (s *stateSyncManager) initTracker() error {
 		ethgo.Address(s.config.stateSenderAddr),
 		s,
 		s.config.numBlockConfirmations,
+		s.config.stateSenderStartBlock,
 		s.logger)
 
 	go func() {

consensus/polybft/statesyncrelayer/state_sync_relayer.go

@@ -24,14 +24,15 @@ import (
 var commitEvent = contractsapi.StateReceiver.Abi.Events["NewCommitment"]
 
 type StateSyncRelayer struct {
-	dataDir           string
-	rpcEndpoint       string
-	stateReceiverAddr ethgo.Address
-	logger            hcf.Logger
-	client            *jsonrpc.Client
-	txRelayer         txrelayer.TxRelayer
-	key               ethgo.Key
-	closeCh           chan struct{}
+	dataDir                string
+	rpcEndpoint            string
+	stateReceiverAddr      ethgo.Address
+	eventTrackerStartBlock uint64
+	logger                 hcf.Logger
+	client                 *jsonrpc.Client
+	txRelayer              txrelayer.TxRelayer
+	key                    ethgo.Key
+	closeCh                chan struct{}
 }
 
 func sanitizeRPCEndpoint(rpcEndpoint string) string {
@@ -51,6 +52,7 @@ func NewRelayer(
 	dataDir string,
 	rpcEndpoint string,
 	stateReceiverAddr ethgo.Address,
+	stateReceiverTrackerStartBlock uint64,
 	logger hcf.Logger,
 	key ethgo.Key,
 ) *StateSyncRelayer {
@@ -70,14 +72,15 @@ func NewRelayer(
 	}
 
 	return &StateSyncRelayer{
-		dataDir:           dataDir,
-		rpcEndpoint:       endpoint,
-		stateReceiverAddr: stateReceiverAddr,
-		logger:            logger,
-		client:            client,
-		txRelayer:         txRelayer,
-		key:               key,
-		closeCh:           make(chan struct{}),
+		dataDir:                dataDir,
+		rpcEndpoint:            endpoint,
+		stateReceiverAddr:      stateReceiverAddr,
+		logger:                 logger,
+		client:                 client,
+		txRelayer:              txRelayer,
+		key:                    key,
+		closeCh:                make(chan struct{}),
+		eventTrackerStartBlock: stateReceiverTrackerStartBlock,
 	}
 }
 
@@ -88,6 +91,7 @@ func (r *StateSyncRelayer) Start() error {
 		r.stateReceiverAddr,
 		r,
 		0, // sidechain (Polygon POS) is instant finality, so no need to wait
+		r.eventTrackerStartBlock,
 		r.logger,
 	)
 

consensus/polybft/statesyncrelayer/state_sync_relayer_test.go

@@ -115,7 +115,7 @@ func TestStateSyncRelayer_Stop(t *testing.T) {
 	key, err := wallet.GenerateKey()
 	require.NoError(t, err)
 
-	r := NewRelayer("test-chain-1", "http://127.0.0.1:8545", ethgo.Address(contracts.StateReceiverContract), hclog.NewNullLogger(), key)
+	r := NewRelayer("test-chain-1", "http://127.0.0.1:8545", ethgo.Address(contracts.StateReceiverContract), 0, hclog.NewNullLogger(), key)
 
 	require.NotPanics(t, func() { r.Stop() })
 }

server/server.go

@@ -15,6 +15,7 @@ import (
 	"github.com/0xPolygon/polygon-edge/blockchain"
 	"github.com/0xPolygon/polygon-edge/chain"
 	"github.com/0xPolygon/polygon-edge/consensus"
+	"github.com/0xPolygon/polygon-edge/consensus/polybft"
 	bls "github.com/0xPolygon/polygon-edge/consensus/polybft/signer"
 	"github.com/0xPolygon/polygon-edge/consensus/polybft/statesyncrelayer"
 	"github.com/0xPolygon/polygon-edge/consensus/polybft/wallet"
@@ -459,10 +460,21 @@ func (s *Server) setupRelayer() error {
 		return fmt.Errorf("failed to create account from secret: %w", err)
 	}
 
+	polyBFTConfig, err := polybft.GetPolyBFTConfig(s.config.Chain)
+	if err != nil {
+		return fmt.Errorf("failed to extract polybft config: %w", err)
+	}
+
+	trackerStartBlockConfig := map[types.Address]uint64{}
+	if polyBFTConfig.Bridge != nil {
+		trackerStartBlockConfig = polyBFTConfig.Bridge.EventTrackerStartBlocks
+	}
+
 	relayer := statesyncrelayer.NewRelayer(
 		s.config.DataDir,
 		s.config.JSONRPC.JSONRPCAddr.String(),
 		ethgo.Address(contracts.StateReceiverContract),
+		trackerStartBlockConfig[contracts.StateReceiverContract],
 		s.logger.Named("relayer"),
 		wallet.NewEcdsaSigner(wallet.NewKey(account, bls.DomainCheckpointManager)),
 	)

tracker/event_tracker.go

@@ -17,6 +17,7 @@ type EventTracker struct {
 	dbPath                string
 	rpcEndpoint           string
 	contractAddr          ethgo.Address
+	startBlock            uint64
 	subscriber            eventSubscription
 	logger                hcf.Logger
 	numBlockConfirmations uint64 // minimal number of child blocks required for the parent block to be considered final
@@ -28,6 +29,7 @@ func NewEventTracker(
 	contractAddr ethgo.Address,
 	subscriber eventSubscription,
 	numBlockConfirmations uint64,
+	startBlock uint64,
 	logger hcf.Logger,
 ) *EventTracker {
 	return &EventTracker{
@@ -36,11 +38,18 @@ func NewEventTracker(
 		contractAddr:          contractAddr,
 		subscriber:            subscriber,
 		numBlockConfirmations: numBlockConfirmations,
+		startBlock:            startBlock,
 		logger:                logger.Named("event_tracker"),
 	}
 }
 
 func (e *EventTracker) Start(ctx context.Context) error {
+	e.logger.Info("Start tracking events",
+		"contract", e.contractAddr,
+		"JSON RPC address", e.rpcEndpoint,
+		"num block confirmations", e.numBlockConfirmations,
+		"start block", e.startBlock)
+
 	provider, err := jsonrpc.NewClient(e.rpcEndpoint)
 	if err != nil {
 		return err
@@ -51,9 +60,6 @@ func (e *EventTracker) Start(ctx context.Context) error {
 		return err
 	}
 
-	e.logger.Info("Start tracking events",
-		"num block confirmations", e.numBlockConfirmations, "contract address", e.contractAddr)
-
 	tt, err := tracker.NewTracker(provider.Eth(),
 		tracker.WithBatchSize(10),
 		tracker.WithStore(store),
@@ -62,6 +68,7 @@ func (e *EventTracker) Start(ctx context.Context) error {
 			Address: []ethgo.Address{
 				e.contractAddr,
 			},
+			Start: e.startBlock,
 		}),
 	)
 	if err != nil {