forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-31 00:53:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Merge pull request #66 from maticnetwork/mew-deposits

Browse source 
alter the fetch logic a bit
This commit is contained in:
Jaynti Kanani 2020-05-20 16:30:29 +05:30 committed by GitHub
commit 7fd50a84e6
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 187 additions and 127 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

89
consensus/bor/bor.go
View file

@ -62,7 +62,6 @@ 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
@ -221,7 +220,7 @@ type Bor struct {
lock sync.RWMutex // Protects the signer fields
ethAPI *ethapi.PublicBlockChainAPI
genesisContractsClient *GenesisContractsClient
GenesisContractsClient *GenesisContractsClient
validatorSetABI abi.ABI
stateReceiverABI abi.ABI
HeimdallClient IHeimdallClient
@ -263,7 +262,7 @@ func New(
signatures: signatures,
validatorSetABI: vABI,
stateReceiverABI: sABI,
genesisContractsClient: genesisContractsClient,
GenesisContractsClient: genesisContractsClient,
HeimdallClient: heimdallClient,
}
@ -1076,57 +1075,26 @@ func (c *Bor) CommitStates(
chain chainContext,
) error {
number := header.Number.Uint64()
lastSync, err := c.genesisContractsClient.LastStateSyncTime(number - 1)
_lastStateID, err := c.GenesisContractsClient.LastStateId(number - 1)
if err != nil {
return err
}
from := lastSync.Add(time.Second * 1) // querying the interval [from, to)
to := time.Unix(int64(chain.Chain.GetHeaderByNumber(number-c.config.Sprint).Time), 0)
if !from.Before(to) {
return nil
}
lastStateID := _lastStateID.Uint64()
log.Info(
"Fetching state updates from Heimdall",
"from", from.Format(time.RFC3339),
"fromID", lastStateID+1,
"to", to.Format(time.RFC3339))
page := 1
eventRecords := make([]*EventRecordWithTime, 0)
for {
queryParams := fmt.Sprintf("from-time=%d&to-time=%d&page=%d&limit=%d", from.Unix(), to.Unix(), page, stateFetchLimit)
log.Info("Fetching state sync events", "queryParams", queryParams)
response, err := c.HeimdallClient.FetchWithRetry("clerk/event-record/list", queryParams)
if err != nil {
return err
}
var _eventRecords []*EventRecordWithTime
if response.Result == nil { // status 204
break
}
if err := json.Unmarshal(response.Result, &_eventRecords); err != nil {
return err
}
eventRecords = append(eventRecords, _eventRecords...)
if len(_eventRecords) < stateFetchLimit {
break
}
page++
}
sort.SliceStable(eventRecords, func(i, j int) bool {
return eventRecords[i].ID < eventRecords[j].ID
})
eventRecords, err := c.HeimdallClient.FetchStateSyncEvents(lastStateID+1, to.Unix())
chainID := c.chainConfig.ChainID.String()
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)))
if eventRecord.ID <= lastStateID {
continue
}
if err := validateEventRecord(eventRecord, number, to, lastStateID, chainID); err != nil {
log.Error(err.Error())
break
}
@ -1140,17 +1108,18 @@ func (c *Bor) CommitStates(
c.stateDataFeed.Send(core.NewStateChangeEvent{StateData: &stateData})
}()
if err := c.genesisContractsClient.CommitState(eventRecord, state, header, chain); err != nil {
if err := c.GenesisContractsClient.CommitState(eventRecord, state, header, chain); err != nil {
return err
}
lastStateID++
}
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}
func validateEventRecord(eventRecord *EventRecordWithTime, number uint64, to time.Time, lastStateID uint64, chainID string) error {
// event id should be sequential and event.Time should lie in the range [from, to)
if lastStateID+1 != eventRecord.ID || eventRecord.ChainID != chainID || !eventRecord.Time.Before(to) {
return &InvalidStateReceivedError{number, lastStateID, &to, eventRecord}
}
return nil
}
@ -1164,28 +1133,6 @@ func (c *Bor) SetHeimdallClient(h IHeimdallClient) {
c.HeimdallClient = h
}
func isProposeSpanAction(tx *types.Transaction, validatorContract string) bool {
// keccak256('proposeSpan()').slice(0, 4)
proposeSpanSig, _ := hex.DecodeString("4b0e4d17")
if tx.Data() == nil || len(tx.Data()) < 4 {
return false
}
return bytes.Compare(proposeSpanSig, tx.Data()[:4]) == 0 &&
tx.To().String() == validatorContract
}
func isProposeStateAction(tx *types.Transaction, stateReceiverContract string) bool {
// keccak256('proposeState(uint256)').slice(0, 4)
proposeStateSig, _ := hex.DecodeString("ede01f17")
if tx.Data() == nil || len(tx.Data()) < 4 {
return false
}
return bytes.Compare(proposeStateSig, tx.Data()[:4]) == 0 &&
tx.To().String() == stateReceiverContract
}
//
// Private methods
//

130
consensus/bor/bor_test/bor_test.go
View file

@ -3,20 +3,17 @@ package bortest
import (
"encoding/hex"
"encoding/json"
"fmt"
"math/big"
"testing"
"time"
"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/consensus/bor"
"github.com/maticnetwork/bor/core/rawdb"
"github.com/maticnetwork/bor/core/types"
"github.com/maticnetwork/bor/crypto"
"github.com/maticnetwork/bor/mocks"
)
@ -36,7 +33,7 @@ func TestInsertingSpanSizeBlocks(t *testing.T) {
db := init.ethereum.ChainDb()
block := init.genesis.ToBlock(db)
to := int64(block.Header().Time)
// 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 <= spanSize; i++ {
@ -45,7 +42,6 @@ func TestInsertingSpanSizeBlocks(t *testing.T) {
}
assert.True(t, h.AssertCalled(t, "FetchWithRetry", spanPath, ""))
assert.True(t, h.AssertCalled(t, "FetchWithRetry", clerkPath, fmt.Sprintf(clerkQueryParams, 1, to, 1)))
validators, err := _bor.GetCurrentValidators(sprintSize, spanSize) // check validator set at the first block of new span
if err != nil {
t.Fatalf("%s", err)
@ -80,39 +76,82 @@ func TestFetchStateSyncEvents(t *testing.T) {
h.On("FetchWithRetry", spanPath, "").Return(res, nil)
// B.2 Mock State Sync events
// read heimdall api response from file
res = stateSyncEventsPayload(t)
var _eventRecords []*bor.EventRecordWithTime
if err := json.Unmarshal(res.Result, &_eventRecords); err != nil {
t.Fatalf("%s", err)
}
// 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
fromID := uint64(1)
// at # sprintSize, events are fetched for [fromID, (block-sprint).Time)
to := int64(chain.GetHeaderByNumber(0).Time)
page := 1
query1Params := fmt.Sprintf(clerkQueryParams, from, to, page)
h.On("FetchWithRetry", clerkPath, query1Params).Return(&response, nil)
eventCount := 50
page = 2
query2Params := fmt.Sprintf(clerkQueryParams, from, to, page)
h.On("FetchWithRetry", clerkPath, query2Params).Return(&bor.ResponseWithHeight{}, nil)
sample := getSampleEventRecord(t)
sample.Time = time.Unix(to-int64(eventCount+1), 0) // last event.Time will be just < to
eventRecords := generateFakeStateSyncEvents(sample, eventCount)
h.On("FetchStateSyncEvents", fromID, to).Return(eventRecords, 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))
assert.True(t, h.AssertCalled(t, "FetchStateSyncEvents", fromID, to))
}
func TestFetchStateSyncEvents_2(t *testing.T) {
init := buildEthereumInstance(t, rawdb.NewMemoryDatabase())
chain := init.ethereum.BlockChain()
engine := init.ethereum.Engine()
_bor := engine.(*bor.Bor)
// Mock /bor/span/1
res, _ := loadSpanFromFile(t)
h := &mocks.IHeimdallClient{}
h.On("FetchWithRetry", spanPath, "").Return(res, nil)
// Mock State Sync events
// at # sprintSize, events are fetched for [fromID, (block-sprint).Time)
fromID := uint64(1)
to := int64(chain.GetHeaderByNumber(0).Time)
sample := getSampleEventRecord(t)
// First query will be from [id=1, (block-sprint).Time]
// Insert 5 events in this time range
eventRecords := []*bor.EventRecordWithTime{
buildStateEvent(sample, 1, 3), // id = 1, time = 1
buildStateEvent(sample, 2, 1), // id = 2, time = 3
buildStateEvent(sample, 3, 2), // id = 3, time = 2
// event with id 5 is missing
buildStateEvent(sample, 4, 5), // id = 4, time = 5
buildStateEvent(sample, 6, 4), // id = 6, time = 4
}
h.On("FetchStateSyncEvents", fromID, to).Return(eventRecords, nil)
_bor.SetHeimdallClient(h)
// Insert blocks for 0th sprint
db := init.ethereum.ChainDb()
block := init.genesis.ToBlock(db)
for i := uint64(1); i <= sprintSize; i++ {
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, "FetchStateSyncEvents", fromID, to))
lastStateID, _ := _bor.GenesisContractsClient.LastStateId(sprintSize)
// state 6 was not written
assert.Equal(t, uint64(4), lastStateID.Uint64())
//
fromID = uint64(5)
to = int64(chain.GetHeaderByNumber(sprintSize).Time)
eventRecords = []*bor.EventRecordWithTime{
buildStateEvent(sample, 5, 7),
buildStateEvent(sample, 6, 4),
}
h.On("FetchStateSyncEvents", fromID, to).Return(eventRecords, nil)
for i := sprintSize + 1; i <= spanSize; i++ {
block = buildNextBlock(t, _bor, chain, block, nil, init.genesis.Config.Bor)
insertNewBlock(t, chain, block)
}
assert.True(t, h.AssertCalled(t, "FetchStateSyncEvents", fromID, to))
lastStateID, _ = _bor.GenesisContractsClient.LastStateId(spanSize)
assert.Equal(t, uint64(6), lastStateID.Uint64())
}
func TestOutOfTurnSigning(t *testing.T) {
@ -191,15 +230,17 @@ func getMockedHeimdallClient(t *testing.T) (*mocks.IHeimdallClient, *bor.Heimdal
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)
h.On(
"FetchStateSyncEvents",
mock.AnythingOfType("uint64"),
mock.AnythingOfType("int64")).Return([]*bor.EventRecordWithTime{getSampleEventRecord(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()
event.ID = 1
events[0] = &bor.EventRecordWithTime{}
*events[0] = event
for i := 1; i < count; i++ {
@ -210,3 +251,20 @@ func generateFakeStateSyncEvents(sample *bor.EventRecordWithTime, count int) []*
}
return events
}
func buildStateEvent(sample *bor.EventRecordWithTime, id uint64, timeStamp int64) *bor.EventRecordWithTime {
event := *sample
event.ID = id
event.Time = time.Unix(timeStamp, 0)
return &event
}
func getSampleEventRecord(t *testing.T) *bor.EventRecordWithTime {
res := stateSyncEventsPayload(t)
var _eventRecords []*bor.EventRecordWithTime
if err := json.Unmarshal(res.Result, &_eventRecords); err != nil {
t.Fatalf("%s", err)
}
_eventRecords[0].Time = time.Unix(1, 0)
return _eventRecords[0]
}

8
consensus/bor/bor_test/genesis.json
View file

File diff suppressed because one or more lines are too long

16
consensus/bor/errors.go
View file

@ -126,18 +126,18 @@ func (e *WrongDifficultyError) Error() string {
}
type InvalidStateReceivedError struct {
Number uint64
From *time.Time
To *time.Time
Event *EventRecordWithTime
Number uint64
LastStateID uint64
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),
"Received invalid event %s at block %d. Requested events until %s. Last state id was %d",
e.Event,
e.Number,
e.To.Format(time.RFC3339),
e.LastStateID,
)
}

13
consensus/bor/genesis_contracts_client.go
View file

File diff suppressed because one or more lines are too long

35
consensus/bor/rest.go
View file

@ -6,11 +6,16 @@ import (
"io/ioutil"
"net/http"
"net/url"
"sort"
"time"
"github.com/maticnetwork/bor/log"
)
var (
stateFetchLimit = 50
)
// ResponseWithHeight defines a response object type that wraps an original
// response with a height.
type ResponseWithHeight struct {
@ -21,6 +26,7 @@ type ResponseWithHeight struct {
type IHeimdallClient interface {
Fetch(path string, query string) (*ResponseWithHeight, error)
FetchWithRetry(path string, query string) (*ResponseWithHeight, error)
FetchStateSyncEvents(fromID uint64, to int64) ([]*EventRecordWithTime, error)
}
type HeimdallClient struct {
@ -38,6 +44,35 @@ func NewHeimdallClient(urlString string) (*HeimdallClient, error) {
return h, nil
}
func (h *HeimdallClient) FetchStateSyncEvents(fromID uint64, to int64) ([]*EventRecordWithTime, error) {
eventRecords := make([]*EventRecordWithTime, 0)
for {
queryParams := fmt.Sprintf("from-id=%d&to-time=%d&limit=%d", fromID, to, stateFetchLimit)
log.Info("Fetching state sync events", "queryParams", queryParams)
response, err := h.FetchWithRetry("clerk/event-record/list", queryParams)
if err != nil {
return nil, err
}
var _eventRecords []*EventRecordWithTime
if response.Result == nil { // status 204
break
}
if err := json.Unmarshal(response.Result, &_eventRecords); err != nil {
return nil, err
}
eventRecords = append(eventRecords, _eventRecords...)
if len(_eventRecords) < stateFetchLimit {
break
}
fromID += uint64(stateFetchLimit)
}
sort.SliceStable(eventRecords, func(i, j int) bool {
return eventRecords[i].ID < eventRecords[j].ID
})
return eventRecords, nil
}
// Fetch fetches response from heimdall
func (h *HeimdallClient) Fetch(rawPath string, rawQuery string) (*ResponseWithHeight, error) {
u, err := url.Parse(h.urlString)

23
mocks/IHeimdallClient.go
View file

@ -35,6 +35,29 @@ func (_m *IHeimdallClient) Fetch(path string, query string) (*bor.ResponseWithHe
return r0, r1
}
// FetchStateSyncEvents provides a mock function with given fields: fromID, to
func (_m *IHeimdallClient) FetchStateSyncEvents(fromID uint64, to int64) ([]*bor.EventRecordWithTime, error) {
ret := _m.Called(fromID, to)
var r0 []*bor.EventRecordWithTime
if rf, ok := ret.Get(0).(func(uint64, int64) []*bor.EventRecordWithTime); ok {
r0 = rf(fromID, to)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).([]*bor.EventRecordWithTime)
}
}
var r1 error
if rf, ok := ret.Get(1).(func(uint64, int64) error); ok {
r1 = rf(fromID, to)
} else {
r1 = ret.Error(1)
}
return r0, r1
}
// FetchWithRetry provides a mock function with given fields: path, query
func (_m *IHeimdallClient) FetchWithRetry(path string, query string) (*bor.ResponseWithHeight, error) {
ret := _m.Called(path, query)