Merge pull request #64 from maticnetwork/mew-deposits

new: New fetch State sync logic

consensus/bor/bor.go

@@ -12,7 +12,6 @@ import (
 	"math/big"
 
 	"sort"
-	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -41,9 +40,6 @@ import (
 	"golang.org/x/crypto/sha3"
 )
 
-const validatorsetABI = `[{"constant":true,"inputs":[{"name":"span","type":"uint256"}],"name":"getSpan","outputs":[{"name":"number","type":"uint256"},{"name":"startBlock","type":"uint256"},{"name":"endBlock","type":"uint256"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[{"name":"number","type":"uint256"}],"name":"getBorValidators","outputs":[{"name":"","type":"address[]"},{"name":"","type":"uint256[]"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[{"name":"span","type":"uint256"},{"name":"signer","type":"address"}],"name":"isProducer","outputs":[{"name":"","type":"bool"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":false,"inputs":[{"name":"newSpan","type":"uint256"},{"name":"startBlock","type":"uint256"},{"name":"endBlock","type":"uint256"},{"name":"validatorBytes","type":"bytes"},{"name":"producerBytes","type":"bytes"}],"name":"commitSpan","outputs":[],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":true,"inputs":[{"name":"span","type":"uint256"},{"name":"signer","type":"address"}],"name":"isValidator","outputs":[{"name":"","type":"bool"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":false,"inputs":[],"name":"proposeSpan","outputs":[],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":true,"inputs":[],"name":"currentSpanNumber","outputs":[{"name":"","type":"uint256"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[],"name":"getNextSpan","outputs":[{"name":"number","type":"uint256"},{"name":"startBlock","type":"uint256"},{"name":"endBlock","type":"uint256"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[],"name":"getInitialValidators","outputs":[{"name":"","type":"address[]"},{"name":"","type":"uint256[]"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[],"name":"spanProposalPending","outputs":[{"name":"","type":"bool"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[],"name":"getCurrentSpan","outputs":[{"name":"number","type":"uint256"},{"name":"startBlock","type":"uint256"},{"name":"endBlock","type":"uint256"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[{"name":"number","type":"uint256"}],"name":"getSpanByBlock","outputs":[{"name":"","type":"uint256"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[],"name":"getValidators","outputs":[{"name":"","type":"address[]"},{"name":"","type":"uint256[]"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":false,"inputs":[{"name":"vote","type":"bytes"},{"name":"sigs","type":"bytes"},{"name":"txBytes","type":"bytes"},{"name":"proof","type":"bytes"}],"name":"validateValidatorSet","outputs":[],"payable":false,"stateMutability":"nonpayable","type":"function"}]`
-const stateReceiverABI = `[{"constant":true,"inputs":[{"name":"","type":"uint256"}],"name":"states","outputs":[{"name":"","type":"bool"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":false,"inputs":[{"name":"recordBytes","type":"bytes"}],"name":"commitState","outputs":[],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":true,"inputs":[],"name":"getPendingStates","outputs":[{"name":"","type":"uint256[]"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[],"name":"SYSTEM_ADDRESS","outputs":[{"name":"","type":"address"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[],"name":"validatorSet","outputs":[{"name":"","type":"address"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":false,"inputs":[{"name":"vote","type":"bytes"},{"name":"sigs","type":"bytes"},{"name":"txBytes","type":"bytes"},{"name":"proof","type":"bytes"}],"name":"validateValidatorSet","outputs":[],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":true,"inputs":[],"name":"isValidatorSetContract","outputs":[{"name":"","type":"bool"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":false,"inputs":[{"name":"stateId","type":"uint256"}],"name":"proposeState","outputs":[],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":true,"inputs":[{"name":"signer","type":"address"}],"name":"isProducer","outputs":[{"name":"","type":"bool"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[{"name":"signer","type":"address"}],"name":"isValidator","outputs":[{"name":"","type":"bool"}],"payable":false,"stateMutability":"view","type":"function"}]`
-
 const (
 	checkpointInterval = 1024 // Number of blocks after which to save the vote snapshot to the database
 	inmemorySnapshots  = 128  // Number of recent vote snapshots to keep in memory
@@ -66,6 +62,7 @@ var (
 
 	validatorHeaderBytesLength = common.AddressLength + 20 // address + power
 	systemAddress              = common.HexToAddress("0xffffFFFfFFffffffffffffffFfFFFfffFFFfFFfE")
+	stateFetchLimit            = 50
 )
 
 // Various error messages to mark blocks invalid. These should be private to
@@ -223,10 +220,11 @@ type Bor struct {
 	signFn SignerFn       // Signer function to authorize hashes with
 	lock   sync.RWMutex   // Protects the signer fields
 
-	ethAPI           *ethapi.PublicBlockChainAPI
+	ethAPI                 *ethapi.PublicBlockChainAPI
-	validatorSetABI  abi.ABI
+	genesisContractsClient *GenesisContractsClient
-	stateReceiverABI abi.ABI
+	validatorSetABI        abi.ABI
-	HeimdallClient   IHeimdallClient
+	stateReceiverABI       abi.ABI
+	HeimdallClient         IHeimdallClient
 
 	stateDataFeed event.Feed
 	scope         event.SubscriptionScope
@@ -255,17 +253,18 @@ func New(
 	vABI, _ := abi.JSON(strings.NewReader(validatorsetABI))
 	sABI, _ := abi.JSON(strings.NewReader(stateReceiverABI))
 	heimdallClient, _ := NewHeimdallClient(heimdallURL)
-
+	genesisContractsClient := NewGenesisContractsClient(chainConfig, borConfig.ValidatorContract, borConfig.StateReceiverContract, ethAPI)
 	c := &Bor{
-		chainConfig:      chainConfig,
+		chainConfig:            chainConfig,
-		config:           borConfig,
+		config:                 borConfig,
-		db:               db,
+		db:                     db,
-		ethAPI:           ethAPI,
+		ethAPI:                 ethAPI,
-		recents:          recents,
+		recents:                recents,
-		signatures:       signatures,
+		signatures:             signatures,
-		validatorSetABI:  vABI,
+		validatorSetABI:        vABI,
-		stateReceiverABI: sABI,
+		stateReceiverABI:       sABI,
-		HeimdallClient:   heimdallClient,
+		genesisContractsClient: genesisContractsClient,
+		HeimdallClient:         heimdallClient,
 	}
 
 	return c
@@ -654,9 +653,7 @@ func (c *Bor) Prepare(chain consensus.ChainReader, header *types.Header) error {
 // Finalize implements consensus.Engine, ensuring no uncles are set, nor block
 // rewards given.
 func (c *Bor) Finalize(chain consensus.ChainReader, header *types.Header, state *state.StateDB, txs []*types.Transaction, uncles []*types.Header) {
-	// commit span
 	headerNumber := header.Number.Uint64()
-
 	if headerNumber%c.config.Sprint == 0 {
 		cx := chainContext{Chain: chain, Bor: c}
 		// check and commit span
@@ -664,6 +661,7 @@ func (c *Bor) Finalize(chain consensus.ChainReader, header *types.Header, state
 			log.Error("Error while committing span", "error", err)
 			return
 		}
+
 		// commit statees
 		if err := c.CommitStates(state, header, cx); err != nil {
 			log.Error("Error while committing states", "error", err)
@@ -679,8 +677,8 @@ func (c *Bor) Finalize(chain consensus.ChainReader, header *types.Header, state
 // FinalizeAndAssemble implements consensus.Engine, ensuring no uncles are set,
 // nor block rewards given, and returns the final block.
 func (c *Bor) FinalizeAndAssemble(chain consensus.ChainReader, header *types.Header, state *state.StateDB, txs []*types.Transaction, uncles []*types.Header, receipts []*types.Receipt) (*types.Block, error) {
-	// commit span
+	headerNumber := header.Number.Uint64()
-	if header.Number.Uint64()%c.config.Sprint == 0 {
+	if headerNumber%c.config.Sprint == 0 {
 		cx := chainContext{Chain: chain, Bor: c}
 
 		// check and commit span
@@ -693,7 +691,7 @@ func (c *Bor) FinalizeAndAssemble(chain consensus.ChainReader, header *types.Hea
 		// commit statees
 		if err := c.CommitStates(state, header, cx); err != nil {
 			log.Error("Error while committing states", "error", err)
-			// return nil, err
+			return nil, err
 		}
 	}
 
@@ -827,38 +825,6 @@ func (c *Bor) Close() error {
 	return nil
 }
 
-// Checks if "force" proposeSpan has been set
-func (c *Bor) isSpanPending(snapshotNumber uint64) (bool, error) {
-	blockNr := rpc.BlockNumber(snapshotNumber)
-	method := "spanProposalPending"
-
-	// get packed data
-	data, err := c.validatorSetABI.Pack(method)
-	if err != nil {
-		log.Error("Unable to pack tx for spanProposalPending", "error", err)
-		return false, err
-	}
-
-	msgData := (hexutil.Bytes)(data)
-	toAddress := common.HexToAddress(c.config.ValidatorContract)
-	gas := (hexutil.Uint64)(uint64(math.MaxUint64 / 2))
-	result, err := c.ethAPI.Call(context.Background(), ethapi.CallArgs{
-		Gas:  &gas,
-		To:   &toAddress,
-		Data: &msgData,
-	}, blockNr)
-	if err != nil {
-		return false, err
-	}
-
-	var ret0 = new(bool)
-	if err := c.validatorSetABI.Unpack(ret0, method, result); err != nil {
-		return false, err
-	}
-
-	return *ret0, nil
-}
-
 // GetCurrentSpan get current span from contract
 func (c *Bor) GetCurrentSpan(snapshotNumber uint64) (*Span, error) {
 	// block
@@ -963,37 +929,14 @@ func (c *Bor) checkAndCommitSpan(
 	chain core.ChainContext,
 ) error {
 	headerNumber := header.Number.Uint64()
-	pending := false
+	span, err := c.GetCurrentSpan(headerNumber - 1)
-	var span *Span = nil
+	if err != nil {
-	errors := make(chan error)
+		return err
-	go func() {
-		var err error
-		pending, err = c.isSpanPending(headerNumber - 1)
-		errors <- err
-	}()
-
-	go func() {
-		var err error
-		span, err = c.GetCurrentSpan(headerNumber - 1)
-		errors <- err
-	}()
-
-	var err error
-	for i := 0; i < 2; i++ {
-		err = <-errors
-		if err != nil {
-			close(errors)
-			return err
-		}
 	}
-	close(errors)
+	if c.needToCommitSpan(span, headerNumber) {
-
-	// commit span if there is new span pending or span is ending or end block is not set
-	if pending || c.needToCommitSpan(span, headerNumber) {
 		err := c.fetchAndCommitSpan(span.ID+1, state, header, chain)
 		return err
 	}
-
 	return nil
 }
 
@@ -1022,7 +965,7 @@ func (c *Bor) fetchAndCommitSpan(
 	header *types.Header,
 	chain core.ChainContext,
 ) error {
-	response, err := c.HeimdallClient.FetchWithRetry("bor", "span", strconv.FormatUint(newSpanID, 10))
+	response, err := c.HeimdallClient.FetchWithRetry(fmt.Sprintf("bor/span/%d", newSpanID), "")
 
 	if err != nil {
 		return err
@@ -1130,83 +1073,85 @@ func (c *Bor) GetPendingStateProposals(snapshotNumber uint64) ([]*big.Int, error
 func (c *Bor) CommitStates(
 	state *state.StateDB,
 	header *types.Header,
-	chain core.ChainContext,
+	chain chainContext,
 ) error {
-	// get pending state proposals
+	number := header.Number.Uint64()
-	stateIds, err := c.GetPendingStateProposals(header.Number.Uint64() - 1)
+	lastSync, err := c.genesisContractsClient.LastStateSyncTime(number - 1)
 	if err != nil {
 		return err
 	}
-
+	from := lastSync.Add(time.Second * 1) // querying the interval [from, to)
-	// state ids
+	to := time.Unix(int64(chain.Chain.GetHeaderByNumber(number-c.config.Sprint).Time), 0)
-	if len(stateIds) > 0 {
+	if !from.Before(to) {
-		log.Debug("Found new proposed states", "numberOfStates", len(stateIds))
+		return nil
 	}
+	log.Info(
+		"Fetching state updates from Heimdall",
+		"from", from.Format(time.RFC3339),
+		"to", to.Format(time.RFC3339))
 
-	method := "commitState"
+	page := 1
-
+	eventRecords := make([]*EventRecordWithTime, 0)
-	// itereate through state ids
+	for {
-	for _, stateID := range stateIds {
+		queryParams := fmt.Sprintf("from-time=%d&to-time=%d&page=%d&limit=%d", from.Unix(), to.Unix(), page, stateFetchLimit)
-		// fetch from heimdall
+		log.Info("Fetching state sync events", "queryParams", queryParams)
-		response, err := c.HeimdallClient.FetchWithRetry("clerk", "event-record", strconv.FormatUint(stateID.Uint64(), 10))
+		response, err := c.HeimdallClient.FetchWithRetry("clerk/event-record/list", queryParams)
 		if err != nil {
 			return err
 		}
-
+		var _eventRecords []*EventRecordWithTime
-		// get event record
+		if response.Result == nil { // status 204
-		var eventRecord EventRecord
+			break
-		if err := json.Unmarshal(response.Result, &eventRecord); err != nil {
+		}
+		if err := json.Unmarshal(response.Result, &_eventRecords); err != nil {
 			return err
 		}
+		eventRecords = append(eventRecords, _eventRecords...)
+		if len(_eventRecords) < stateFetchLimit {
+			break
+		}
+		page++
+	}
 
-		// check if chain id matches with event record
+	sort.SliceStable(eventRecords, func(i, j int) bool {
-		if eventRecord.ChainID != c.chainConfig.ChainID.String() {
+		return eventRecords[i].ID < eventRecords[j].ID
-			return fmt.Errorf(
+	})
-				"Chain id proposed state in span, %s, and bor chain id, %s, doesn't match",
+
-				eventRecord.ChainID,
+	chainID := c.chainConfig.ChainID.String()
-				c.chainConfig.ChainID,
+	for _, eventRecord := range eventRecords {
-			)
+		// validateEventRecord checks whether an event lies in the specified time range
+		// since the events are sorted by time and if it turns out that event i lies outside the time range,
+		// it would mean all subsequent events lie outside of the time range. Hence we don't probe any further and break the loop
+		if err := validateEventRecord(eventRecord, number, from, to, chainID); err != nil {
+			log.Error(
+				fmt.Sprintf(
+					"Received event %s does not lie in the time range, from %s, to %s",
+					eventRecord, from.Format(time.RFC3339), to.Format(time.RFC3339)))
+			break
 		}
 
-		log.Info("→ committing new state",
-			"id", eventRecord.ID,
-			"contract", eventRecord.Contract,
-			"data", hex.EncodeToString(eventRecord.Data),
-			"txHash", eventRecord.TxHash,
-			"chainID", eventRecord.ChainID,
-		)
 		stateData := types.StateData{
 			Did:      eventRecord.ID,
 			Contract: eventRecord.Contract,
 			Data:     hex.EncodeToString(eventRecord.Data),
 			TxHash:   eventRecord.TxHash,
 		}
-
 		go func() {
 			c.stateDataFeed.Send(core.NewStateChangeEvent{StateData: &stateData})
 		}()
 
-		recordBytes, err := rlp.EncodeToBytes(eventRecord)
+		if err := c.genesisContractsClient.CommitState(eventRecord, state, header, chain); err != nil {
-		if err != nil {
-			return err
-		}
-
-		// get packed data for commit state
-		data, err := c.stateReceiverABI.Pack(method, recordBytes)
-		if err != nil {
-			log.Error("Unable to pack tx for commitState", "error", err)
-			return err
-		}
-
-		// get system message
-		msg := getSystemMessage(common.HexToAddress(c.config.StateReceiverContract), data)
-
-		// apply message
-		if err := applyMessage(msg, state, header, c.chainConfig, chain); err != nil {
 			return err
 		}
 	}
+	return nil
+}
 
+func validateEventRecord(eventRecord *EventRecordWithTime, number uint64, from, to time.Time, chainID string) error {
+	// event should lie in the range [from, to)
+	if eventRecord.ChainID != chainID || eventRecord.Time.Before(from) || !eventRecord.Time.Before(to) {
+		return &InvalidStateReceivedError{number, &from, &to, eventRecord}
+	}
 	return nil
 }
 
@@ -1219,26 +1164,6 @@ func (c *Bor) SetHeimdallClient(h IHeimdallClient) {
 	c.HeimdallClient = h
 }
 
-func (c *Bor) IsValidatorAction(chain consensus.ChainReader, from common.Address, tx *types.Transaction) bool {
-	header := chain.CurrentHeader()
-	validators, err := c.GetCurrentValidators(header.Number.Uint64(), header.Number.Uint64()+1)
-	if err != nil {
-		log.Error("Failed fetching snapshot", err)
-		return false
-	}
-
-	isValidator := false
-	for _, validator := range validators {
-		if bytes.Compare(validator.Address.Bytes(), from.Bytes()) == 0 {
-			isValidator = true
-			break
-		}
-	}
-
-	return isValidator && (isProposeSpanAction(tx, chain.Config().Bor.ValidatorContract) ||
-		isProposeStateAction(tx, chain.Config().Bor.StateReceiverContract))
-}
-
 func isProposeSpanAction(tx *types.Transaction, validatorContract string) bool {
 	// keccak256('proposeSpan()').slice(0, 4)
 	proposeSpanSig, _ := hex.DecodeString("4b0e4d17")

consensus/bor/bor_test/bor_test.go

@@ -2,45 +2,51 @@ package bortest
 
 import (
 	"encoding/hex"
+	"encoding/json"
+	"fmt"
 	"math/big"
 	"testing"
+	"time"
 
-	"github.com/maticnetwork/bor/common"
 	"github.com/maticnetwork/bor/consensus/bor"
 	"github.com/maticnetwork/bor/core/rawdb"
+
 	"github.com/maticnetwork/bor/crypto"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 
 	"github.com/maticnetwork/bor/core/types"
 
 	"github.com/maticnetwork/bor/mocks"
 )
 
-func TestCommitSpan(t *testing.T) {
+var (
+	spanPath         = "bor/span/1"
+	clerkPath        = "clerk/event-record/list"
+	clerkQueryParams = "from-time=%d&to-time=%d&page=%d&limit=50"
+)
+
+func TestInsertingSpanSizeBlocks(t *testing.T) {
 	init := buildEthereumInstance(t, rawdb.NewMemoryDatabase())
 	chain := init.ethereum.BlockChain()
 	engine := init.ethereum.Engine()
 	_bor := engine.(*bor.Bor)
-
+	h, heimdallSpan := getMockedHeimdallClient(t)
-	// Mock HeimdallClient.FetchWithRetry to return span data from span.json
-	res, heimdallSpan := loadSpanFromFile(t)
-	h := &mocks.IHeimdallClient{}
-	h.On("FetchWithRetry", "bor", "span", "1").Return(res, nil)
 	_bor.SetHeimdallClient(h)
 
 	db := init.ethereum.ChainDb()
 	block := init.genesis.ToBlock(db)
+	to := int64(block.Header().Time)
+
 	// Insert sprintSize # of blocks so that span is fetched at the start of a new sprint
-	for i := uint64(1); i <= sprintSize; i++ {
+	for i := uint64(1); i <= spanSize; i++ {
 		block = buildNextBlock(t, _bor, chain, block, nil, init.genesis.Config.Bor)
 		insertNewBlock(t, chain, block)
 	}
 
-	// FetchWithRetry is invoked 2 times
+	assert.True(t, h.AssertCalled(t, "FetchWithRetry", spanPath, ""))
-	// 1. bor.FinalizeAndAssemble to prepare a new block when calling buildNextBlock
+	assert.True(t, h.AssertCalled(t, "FetchWithRetry", clerkPath, fmt.Sprintf(clerkQueryParams, 1, to, 1)))
-	// 2. bor.Finalize via(bc.insertChain => bc.processor.Process)
+	validators, err := _bor.GetCurrentValidators(sprintSize, spanSize) // check validator set at the first block of new span
-	assert.True(t, h.AssertCalled(t, "FetchWithRetry", "bor", "span", "1"))
-	validators, err := _bor.GetCurrentValidators(sprintSize, 256) // new span starts at 256
 	if err != nil {
 		t.Fatalf("%s", err)
 	}
@@ -52,62 +58,61 @@ func TestCommitSpan(t *testing.T) {
 	}
 }
 
-func TestIsValidatorAction(t *testing.T) {
+func TestFetchStateSyncEvents(t *testing.T) {
 	init := buildEthereumInstance(t, rawdb.NewMemoryDatabase())
 	chain := init.ethereum.BlockChain()
 	engine := init.ethereum.Engine()
 	_bor := engine.(*bor.Bor)
 
-	proposeStateData, _ := hex.DecodeString("ede01f170000000000000000000000000000000000000000000000000000000000000000")
+	// A. Insert blocks for 0th sprint
-	proposeSpanData, _ := hex.DecodeString("4b0e4d17")
-	var tx *types.Transaction
-	tx = types.NewTransaction(
-		0,
-		common.HexToAddress(chain.Config().Bor.StateReceiverContract),
-		big.NewInt(0), 0, big.NewInt(0),
-		proposeStateData,
-	)
-	assert.True(t, _bor.IsValidatorAction(chain, addr, tx))
-
-	tx = types.NewTransaction(
-		0,
-		common.HexToAddress(chain.Config().Bor.ValidatorContract),
-		big.NewInt(0), 0, big.NewInt(0),
-		proposeSpanData,
-	)
-	assert.True(t, _bor.IsValidatorAction(chain, addr, tx))
-
-	res, heimdallSpan := loadSpanFromFile(t)
-	h := &mocks.IHeimdallClient{}
-	h.On("FetchWithRetry", "bor", "span", "1").Return(res, nil)
-	_bor.SetHeimdallClient(h)
-
 	db := init.ethereum.ChainDb()
 	block := init.genesis.ToBlock(db)
-
+	// Insert sprintSize # of blocks so that span is fetched at the start of a new sprint
-	for i := uint64(1); i <= spanSize; i++ {
+	for i := uint64(1); i < sprintSize; i++ {
 		block = buildNextBlock(t, _bor, chain, block, nil, init.genesis.Config.Bor)
 		insertNewBlock(t, chain, block)
 	}
 
-	for _, validator := range heimdallSpan.SelectedProducers {
+	// B. Before inserting 1st block of the next sprint, mock heimdall deps
-		_addr := validator.Address
+	// B.1 Mock /bor/span/1
-		tx = types.NewTransaction(
+	res, _ := loadSpanFromFile(t)
-			0,
+	h := &mocks.IHeimdallClient{}
-			common.HexToAddress(chain.Config().Bor.StateReceiverContract),
+	h.On("FetchWithRetry", spanPath, "").Return(res, nil)
-			big.NewInt(0), 0, big.NewInt(0),
-			proposeStateData,
-		)
-		assert.True(t, _bor.IsValidatorAction(chain, _addr, tx))
 
-		tx = types.NewTransaction(
+	// B.2 Mock State Sync events
-			0,
+	// read heimdall api response from file
-			common.HexToAddress(chain.Config().Bor.ValidatorContract),
+	res = stateSyncEventsPayload(t)
-			big.NewInt(0), 0, big.NewInt(0),
+	var _eventRecords []*bor.EventRecordWithTime
-			proposeSpanData,
+	if err := json.Unmarshal(res.Result, &_eventRecords); err != nil {
-		)
+		t.Fatalf("%s", err)
-		assert.True(t, _bor.IsValidatorAction(chain, _addr, tx))
 	}
+
+	// use that as a sample to generate bor.stateFetchLimit events
+	eventRecords := generateFakeStateSyncEvents(_eventRecords[0], 50)
+	_res, _ := json.Marshal(eventRecords)
+	response := bor.ResponseWithHeight{Height: "0"}
+	if err := json.Unmarshal(_res, &response.Result); err != nil {
+		t.Fatalf("%s", err)
+	}
+
+	// at # sprintSize, events are fetched for the interval [from, (block-sprint).Time)
+	from := 1
+	to := int64(chain.GetHeaderByNumber(0).Time)
+	page := 1
+	query1Params := fmt.Sprintf(clerkQueryParams, from, to, page)
+	h.On("FetchWithRetry", clerkPath, query1Params).Return(&response, nil)
+
+	page = 2
+	query2Params := fmt.Sprintf(clerkQueryParams, from, to, page)
+	h.On("FetchWithRetry", clerkPath, query2Params).Return(&bor.ResponseWithHeight{}, nil)
+	_bor.SetHeimdallClient(h)
+
+	block = buildNextBlock(t, _bor, chain, block, nil, init.genesis.Config.Bor)
+	insertNewBlock(t, chain, block)
+
+	assert.True(t, h.AssertCalled(t, "FetchWithRetry", spanPath, ""))
+	assert.True(t, h.AssertCalled(t, "FetchWithRetry", clerkPath, query1Params))
+	assert.True(t, h.AssertCalled(t, "FetchWithRetry", clerkPath, query2Params))
 }
 
 func TestOutOfTurnSigning(t *testing.T) {
@@ -115,10 +120,7 @@ func TestOutOfTurnSigning(t *testing.T) {
 	chain := init.ethereum.BlockChain()
 	engine := init.ethereum.Engine()
 	_bor := engine.(*bor.Bor)
-
+	h, _ := getMockedHeimdallClient(t)
-	res, _ := loadSpanFromFile(t)
-	h := &mocks.IHeimdallClient{}
-	h.On("FetchWithRetry", "bor", "span", "1").Return(res, nil)
 	_bor.SetHeimdallClient(h)
 
 	db := init.ethereum.ChainDb()
@@ -166,10 +168,7 @@ func TestSignerNotFound(t *testing.T) {
 	chain := init.ethereum.BlockChain()
 	engine := init.ethereum.Engine()
 	_bor := engine.(*bor.Bor)
-
+	h, _ := getMockedHeimdallClient(t)
-	res, _ := loadSpanFromFile(t)
-	h := &mocks.IHeimdallClient{}
-	h.On("FetchWithRetry", "bor", "span", "1").Return(res, nil)
 	_bor.SetHeimdallClient(h)
 
 	db := init.ethereum.ChainDb()
@@ -187,3 +186,27 @@ func TestSignerNotFound(t *testing.T) {
 		*err.(*bor.UnauthorizedSignerError),
 		bor.UnauthorizedSignerError{Number: 0, Signer: addr.Bytes()})
 }
+
+func getMockedHeimdallClient(t *testing.T) (*mocks.IHeimdallClient, *bor.HeimdallSpan) {
+	res, heimdallSpan := loadSpanFromFile(t)
+	h := &mocks.IHeimdallClient{}
+	h.On("FetchWithRetry", "bor/span/1", "").Return(res, nil)
+	h.On("FetchWithRetry", mock.AnythingOfType("string"), mock.AnythingOfType("string")).Return(stateSyncEventsPayload(t), nil)
+	return h, heimdallSpan
+}
+
+func generateFakeStateSyncEvents(sample *bor.EventRecordWithTime, count int) []*bor.EventRecordWithTime {
+	events := make([]*bor.EventRecordWithTime, count)
+	event := *sample
+	event.ID = 0
+	event.Time = time.Now()
+	events[0] = &bor.EventRecordWithTime{}
+	*events[0] = event
+	for i := 1; i < count; i++ {
+		event.ID = uint64(i)
+		event.Time = event.Time.Add(1 * time.Second)
+		events[i] = &bor.EventRecordWithTime{}
+		*events[i] = event
+	}
+	return events
+}

consensus/bor/bor_test/helper.go

@@ -156,6 +156,18 @@ func sign(t *testing.T, header *types.Header, signer []byte) {
 	copy(header.Extra[len(header.Extra)-extraSeal:], sig)
 }
 
+func stateSyncEventsPayload(t *testing.T) *bor.ResponseWithHeight {
+	stateData, err := ioutil.ReadFile("states.json")
+	if err != nil {
+		t.Fatalf("%s", err)
+	}
+	res := &bor.ResponseWithHeight{}
+	if err := json.Unmarshal(stateData, res); err != nil {
+		t.Fatalf("%s", err)
+	}
+	return res
+}
+
 func loadSpanFromFile(t *testing.T) (*bor.ResponseWithHeight, *bor.HeimdallSpan) {
 	spanData, err := ioutil.ReadFile("span.json")
 	if err != nil {

consensus/bor/bor_test/states.json

@@ -0,0 +1,23 @@
+{
+  "height": "0",
+  "result": [
+    {
+      "id": 1,
+      "contract": "0xb55969a6d60413a63291a1de572269875df541e3",
+      "data": "0x00000000000000000000000048aa8d4af32551892fcf08ad63be7dd206d46f6500000000000000000000000048aa8d4af32551892fcf08ad63be7dd206d46f65000000000000000000000000000000000000000000000000000000000000000a0000000000000000000000000000000000000000000000000000000000000014",
+      "tx_hash": "0x7b113e09d98b6d4be1dedfbc0746e34876de767f2cb8b58ff00160a160811dd6",
+      "log_index": 0,
+      "bor_chain_id": "15001",
+      "record_time": "2020-05-15T13:36:38.580995Z"
+    },
+    {
+      "id": 2,
+      "contract": "0xb55969a6d60413a63291a1de572269875df541e3",
+      "data": "0x00000000000000000000000048aa8d4af32551892fcf08ad63be7dd206d46f6500000000000000000000000048aa8d4af32551892fcf08ad63be7dd206d46f65000000000000000000000000000000000000000000000000000000000000000a0000000000000000000000000000000000000000000000000000000000000015",
+      "tx_hash": "0xb72358aff8e4d61f4de97a37a40ddda986c081e0de8036e0a78c4b61b067cba9",
+      "log_index": 0,
+      "bor_chain_id": "15001",
+      "record_time": "2020-05-15T13:42:37.319058Z"
+    }
+  ]
+}

consensus/bor/clerk.go

@@ -1,6 +1,9 @@
 package bor
 
 import (
+	"fmt"
+	"time"
+
 	"github.com/maticnetwork/bor/common"
 	"github.com/maticnetwork/bor/common/hexutil"
 )
@@ -14,3 +17,33 @@ type EventRecord struct {
 	LogIndex uint64         `json:"log_index" yaml:"log_index"`
 	ChainID  string         `json:"bor_chain_id" yaml:"bor_chain_id"`
 }
+
+type EventRecordWithTime struct {
+	EventRecord
+	Time time.Time `json:"record_time" yaml:"record_time"`
+}
+
+// String returns the string representatin of span
+func (e *EventRecordWithTime) String() string {
+	return fmt.Sprintf(
+		"id %v, contract %v, data: %v, txHash: %v, logIndex: %v, chainId: %v, time %s",
+		e.ID,
+		e.Contract.String(),
+		e.Data.String(),
+		e.TxHash.Hex(),
+		e.LogIndex,
+		e.ChainID,
+		e.Time.Format(time.RFC3339),
+	)
+}
+
+func (e *EventRecordWithTime) BuildEventRecord() *EventRecord {
+	return &EventRecord{
+		ID:       e.ID,
+		Contract: e.Contract,
+		Data:     e.Data,
+		TxHash:   e.TxHash,
+		LogIndex: e.LogIndex,
+		ChainID:  e.ChainID,
+	}
+}

consensus/bor/errors.go

@@ -2,6 +2,7 @@ package bor
 
 import (
 	"fmt"
+	"time"
 )
 
 // TotalVotingPowerExceededError is returned when the maximum allowed total voting power is exceeded
@@ -123,3 +124,20 @@ func (e *WrongDifficultyError) Error() string {
 		e.Signer,
 	)
 }
+
+type InvalidStateReceivedError struct {
+	Number uint64
+	From   *time.Time
+	To     *time.Time
+	Event  *EventRecordWithTime
+}
+
+func (e *InvalidStateReceivedError) Error() string {
+	return fmt.Sprintf(
+		"Received event with invalid timestamp at block %d. Requested events from %s to %s. Received %s\n",
+		e.Number,
+		e.From.Format(time.RFC3339),
+		e.To.Format(time.RFC3339),
+		e.Event,
+	)
+}

consensus/bor/rest.go

@@ -6,7 +6,6 @@ import (
 	"io/ioutil"
 	"net/http"
 	"net/url"
-	"path"
 	"time"
 
 	"github.com/maticnetwork/bor/log"
@@ -20,8 +19,8 @@ type ResponseWithHeight struct {
 }
 
 type IHeimdallClient interface {
-	Fetch(paths ...string) (*ResponseWithHeight, error)
+	Fetch(path string, query string) (*ResponseWithHeight, error)
-	FetchWithRetry(paths ...string) (*ResponseWithHeight, error)
+	FetchWithRetry(path string, query string) (*ResponseWithHeight, error)
 }
 
 type HeimdallClient struct {
@@ -39,33 +38,28 @@ func NewHeimdallClient(urlString string) (*HeimdallClient, error) {
 	return h, nil
 }
 
-func (h *HeimdallClient) Fetch(paths ...string) (*ResponseWithHeight, error) {
+// Fetch fetches response from heimdall
+func (h *HeimdallClient) Fetch(rawPath string, rawQuery string) (*ResponseWithHeight, error) {
 	u, err := url.Parse(h.urlString)
 	if err != nil {
 		return nil, err
 	}
 
-	for _, e := range paths {
+	u.Path = rawPath
-		if e != "" {
+	u.RawQuery = rawQuery
-			u.Path = path.Join(u.Path, e)
-		}
-	}
 
 	return h.internalFetch(u)
 }
 
 // FetchWithRetry returns data from heimdall with retry
-func (h *HeimdallClient) FetchWithRetry(paths ...string) (*ResponseWithHeight, error) {
+func (h *HeimdallClient) FetchWithRetry(rawPath string, rawQuery string) (*ResponseWithHeight, error) {
 	u, err := url.Parse(h.urlString)
 	if err != nil {
 		return nil, err
 	}
 
-	for _, e := range paths {
+	u.Path = rawPath
-		if e != "" {
+	u.RawQuery = rawQuery
-			u.Path = path.Join(u.Path, e)
-		}
-	}
 
 	for {
 		res, err := h.internalFetch(u)

consensus/consensus.go

@@ -116,12 +116,6 @@ type Engine interface {
 	Close() error
 }
 
-// Bor is a consensus engine developed by folks at Matic Network
-type Bor interface {
-	Engine
-	IsValidatorAction(chain ChainReader, from common.Address, tx *types.Transaction) bool
-}
-
 // PoW is a consensus engine based on proof-of-work.
 type PoW interface {
 	Engine

core/tx_pool.go

@@ -531,7 +531,6 @@ func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error {
 	// Drop non-local transactions under our own minimal accepted gas price
 	local = local || pool.locals.contains(from) // account may be local even if the transaction arrived from the network
 	if !local &&
-		!pool.chain.Engine().(consensus.Bor).IsValidatorAction(pool.chain.(consensus.ChainReader), from, tx) &&
 		pool.gasPrice.Cmp(tx.GasPrice()) > 0 {
 		return ErrUnderpriced
 	}

mocks/IHeimdallClient.go

@@ -12,19 +12,13 @@ type IHeimdallClient struct {
 	mock.Mock
 }
 
-// Fetch provides a mock function with given fields: paths
+// Fetch provides a mock function with given fields: path, query
-func (_m *IHeimdallClient) Fetch(paths ...string) (*bor.ResponseWithHeight, error) {
+func (_m *IHeimdallClient) Fetch(path string, query string) (*bor.ResponseWithHeight, error) {
-	_va := make([]interface{}, len(paths))
+	ret := _m.Called(path, query)
-	for _i := range paths {
-		_va[_i] = paths[_i]
-	}
-	var _ca []interface{}
-	_ca = append(_ca, _va...)
-	ret := _m.Called(_ca...)
 
 	var r0 *bor.ResponseWithHeight
-	if rf, ok := ret.Get(0).(func(...string) *bor.ResponseWithHeight); ok {
+	if rf, ok := ret.Get(0).(func(string, string) *bor.ResponseWithHeight); ok {
-		r0 = rf(paths...)
+		r0 = rf(path, query)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).(*bor.ResponseWithHeight)
@@ -32,8 +26,8 @@ func (_m *IHeimdallClient) Fetch(paths ...string) (*bor.ResponseWithHeight, erro
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(...string) error); ok {
+	if rf, ok := ret.Get(1).(func(string, string) error); ok {
-		r1 = rf(paths...)
+		r1 = rf(path, query)
 	} else {
 		r1 = ret.Error(1)
 	}
@@ -41,19 +35,13 @@ func (_m *IHeimdallClient) Fetch(paths ...string) (*bor.ResponseWithHeight, erro
 	return r0, r1
 }
 
-// FetchWithRetry provides a mock function with given fields: paths
+// FetchWithRetry provides a mock function with given fields: path, query
-func (_m *IHeimdallClient) FetchWithRetry(paths ...string) (*bor.ResponseWithHeight, error) {
+func (_m *IHeimdallClient) FetchWithRetry(path string, query string) (*bor.ResponseWithHeight, error) {
-	_va := make([]interface{}, len(paths))
+	ret := _m.Called(path, query)
-	for _i := range paths {
-		_va[_i] = paths[_i]
-	}
-	var _ca []interface{}
-	_ca = append(_ca, _va...)
-	ret := _m.Called(_ca...)
 
 	var r0 *bor.ResponseWithHeight
-	if rf, ok := ret.Get(0).(func(...string) *bor.ResponseWithHeight); ok {
+	if rf, ok := ret.Get(0).(func(string, string) *bor.ResponseWithHeight); ok {
-		r0 = rf(paths...)
+		r0 = rf(path, query)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).(*bor.ResponseWithHeight)
@@ -61,8 +49,8 @@ func (_m *IHeimdallClient) FetchWithRetry(paths ...string) (*bor.ResponseWithHei
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(...string) error); ok {
+	if rf, ok := ret.Get(1).(func(string, string) error); ok {
-		r1 = rf(paths...)
+		r1 = rf(path, query)
 	} else {
 		r1 = ret.Error(1)
 	}