chore(chainsync): add unit test for reorgs

Author: ezdac

rolling-shutter/medley/chainsync/client.go

@@ -24,7 +24,7 @@ var noopLogger = &logger.NoopLogger{}
 var ErrServiceNotInstantiated = errors.New("service is not instantiated, pass a handler function option")
 
 type Client struct {
-	client.EthereumClient
+	client.SyncEthereumClient
 	log log.Logger
 
 	options *options
@@ -136,7 +136,7 @@ func (s *Client) BroadcastEonKey(ctx context.Context, eon uint64, eonPubKey []by
 // This value is cached, since it is not expected to change.
 func (s *Client) ChainID(ctx context.Context) (*big.Int, error) {
 	if s.chainID == nil {
-		cid, err := s.EthereumClient.ChainID(ctx)
+		cid, err := s.SyncEthereumClient.ChainID(ctx)
 		if err != nil {
 			return nil, err
 		}

rolling-shutter/medley/chainsync/client/client.go

@@ -9,7 +9,7 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 )
 
-type EthereumClient interface {
+type FullEthereumClient interface {
 	Close()
 	ChainID(ctx context.Context) (*big.Int, error)
 	BlockByHash(ctx context.Context, hash common.Hash) (*types.Block, error)
@@ -45,3 +45,14 @@ type EthereumClient interface {
 	EstimateGas(ctx context.Context, msg ethereum.CallMsg) (uint64, error)
 	SendTransaction(ctx context.Context, tx *types.Transaction) error
 }
+
+type SyncEthereumClient interface {
+	Close()
+	ChainID(ctx context.Context) (*big.Int, error)
+	BlockNumber(ctx context.Context) (uint64, error)
+	HeaderByHash(ctx context.Context, hash common.Hash) (*types.Header, error)
+	HeaderByNumber(ctx context.Context, number *big.Int) (*types.Header, error)
+	SubscribeNewHead(ctx context.Context, ch chan<- *types.Header) (ethereum.Subscription, error)
+	FilterLogs(ctx context.Context, q ethereum.FilterQuery) ([]types.Log, error)
+	SubscribeFilterLogs(ctx context.Context, q ethereum.FilterQuery, ch chan<- types.Log) (ethereum.Subscription, error)
+}

rolling-shutter/medley/chainsync/client/test.go

@@ -0,0 +1,162 @@
+package client
+
+import (
+	"context"
+	"errors"
+	"math/big"
+
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+)
+
+var ErrNotImplemented = errors.New("not implemented")
+
+var _ SyncEthereumClient = &TestClient{}
+
+type TestClient struct {
+	headerChain            []*types.Header
+	latestHeadIndex        int
+	intialProgress         bool
+	latestHeadEmitter      []chan<- *types.Header
+	latestHeadSubscription []*Subscription
+}
+
+func NewSubscription(idx int) *Subscription {
+	return &Subscription{
+		idx: idx,
+		err: make(chan error, 1),
+	}
+}
+
+type Subscription struct {
+	idx int
+	err chan error
+}
+
+func (su *Subscription) Unsubscribe() {
+	// TODO: not implemented yet, but we don't want to panic
+}
+
+func (su *Subscription) Err() <-chan error {
+	return su.err
+}
+
+type TestClientController struct {
+	c *TestClient
+}
+
+func NewTestClient() (*TestClient, *TestClientController) {
+	c := &TestClient{
+		headerChain:     []*types.Header{},
+		latestHeadIndex: 0,
+	}
+	ctrl := &TestClientController{c}
+	return c, ctrl
+}
+
+func (c *TestClientController) ProgressHead() bool {
+	if c.c.latestHeadIndex >= len(c.c.headerChain)-1 {
+		return false
+	}
+	c.c.latestHeadIndex++
+	return true
+}
+
+func (c *TestClientController) EmitEvents(ctx context.Context) error {
+	if len(c.c.latestHeadEmitter) == 0 {
+		return nil
+	}
+	h := c.c.getLatestHeader()
+	for _, em := range c.c.latestHeadEmitter {
+		select {
+		case em <- h:
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+	return nil
+}
+
+func (c *TestClientController) AppendNextHeaders(h ...*types.Header) {
+	c.c.headerChain = append(c.c.headerChain, h...)
+}
+
+func (t *TestClient) ChainID(_ context.Context) (*big.Int, error) {
+	return big.NewInt(42), nil
+}
+
+func (t *TestClient) Close() {
+	// TODO: cleanup
+}
+
+func (t *TestClient) getLatestHeader() *types.Header {
+	if len(t.headerChain) == 0 {
+		return nil
+	}
+	return t.headerChain[t.latestHeadIndex]
+}
+
+func (t *TestClient) searchBlock(f func(*types.Header) bool) *types.Header {
+	for i := t.latestHeadIndex; i >= 0; i-- {
+		h := t.headerChain[i]
+		if f(h) {
+			return h
+		}
+	}
+	return nil
+}
+
+func (t *TestClient) searchBlockByNumber(number *big.Int) *types.Header {
+	return t.searchBlock(
+		func(h *types.Header) bool {
+			return h.Number.Cmp(number) == 0
+		})
+}
+
+func (t *TestClient) searchBlockByHash(hash common.Hash) *types.Header {
+	return t.searchBlock(
+		func(h *types.Header) bool {
+			return hash.Cmp(h.Hash()) == 0
+		})
+}
+
+func (t *TestClient) BlockNumber(_ context.Context) (uint64, error) {
+	return t.getLatestHeader().Nonce.Uint64(), nil
+}
+
+func (t *TestClient) HeaderByHash(_ context.Context, hash common.Hash) (*types.Header, error) {
+	h := t.searchBlockByHash(hash)
+	if h == nil {
+		// TODO: what error?
+	}
+	return h, nil
+}
+
+func (t *TestClient) HeaderByNumber(_ context.Context, number *big.Int) (*types.Header, error) {
+	if number == nil {
+		return t.getLatestHeader(), nil
+	}
+	h := t.searchBlockByNumber(number)
+	if h == nil {
+		// TODO: what error?
+	}
+	return h, nil
+}
+
+func (t *TestClient) SubscribeNewHead(_ context.Context, ch chan<- *types.Header) (ethereum.Subscription, error) {
+	t.latestHeadEmitter = append(t.latestHeadEmitter, ch)
+	su := NewSubscription(len(t.latestHeadSubscription) - 1)
+	t.latestHeadSubscription = append(t.latestHeadSubscription, su)
+	// TODO: unsubscribe and deleting from the array
+	// TODO: filling error promise in the subscription
+	return su, nil
+}
+
+func (t *TestClient) FilterLogs(_ context.Context, q ethereum.FilterQuery) ([]types.Log, error) {
+	panic(ErrNotImplemented)
+}
+
+func (t *TestClient) SubscribeFilterLogs(_ context.Context, q ethereum.FilterQuery, ch chan<- types.Log) (ethereum.Subscription, error) {
+	panic(ErrNotImplemented)
+}

rolling-shutter/medley/chainsync/options.go

@@ -24,7 +24,7 @@ type options struct {
 	keyperSetManagerAddress     *common.Address
 	keyBroadcastContractAddress *common.Address
 	clientURL                   string
-	ethClient                   syncclient.EthereumClient
+	ethClient                   syncclient.FullEthereumClient
 	logger                      log.Logger
 	runner                      service.Runner
 	syncStart                   *number.BlockNumber
@@ -122,7 +122,7 @@ func (o *options) applyHandler(c *Client) error {
 // of shutter clients background workers.
 func (o *options) apply(ctx context.Context, c *Client) error {
 	var (
-		client syncclient.EthereumClient
+		client syncclient.SyncEthereumClient
 		err    error
 	)
 	if o.clientURL != "" {
@@ -132,12 +132,12 @@ func (o *options) apply(ctx context.Context, c *Client) error {
 		}
 	}
 	client = o.ethClient
-	c.EthereumClient = client
+	c.SyncEthereumClient = client
 
 	// the nil passthrough will use "latest" for each call,
 	// but we want to harmonize and fix the sync start to a specific block.
 	if o.syncStart.IsLatest() {
-		latestBlock, err := c.EthereumClient.BlockNumber(ctx)
+		latestBlock, err := c.SyncEthereumClient.BlockNumber(ctx)
 		if err != nil {
 			return errors.Wrap(err, "polling latest block")
 		}
@@ -219,7 +219,7 @@ func WithLogger(l log.Logger) Option {
 	}
 }
 
-func WithClient(client syncclient.EthereumClient) Option {
+func WithClient(client syncclient.FullEthereumClient) Option {
 	return func(o *options) error {
 		o.ethClient = client
 		return nil

rolling-shutter/medley/chainsync/syncer/eonpubkey.go

@@ -17,7 +17,7 @@ import (
 var _ ManualFilterHandler = &EonPubKeySyncer{}
 
 type EonPubKeySyncer struct {
-	Client           client.EthereumClient
+	Client           client.SyncEthereumClient
 	Log              log.Logger
 	KeyBroadcast     *bindings.KeyBroadcastContract
 	KeyperSetManager *bindings.KeyperSetManager

rolling-shutter/medley/chainsync/syncer/keyperset.go

@@ -23,7 +23,7 @@ const channelSize = 10
 var _ ManualFilterHandler = &KeyperSetSyncer{}
 
 type KeyperSetSyncer struct {
-	Client   client.EthereumClient
+	Client   client.FullEthereumClient
 	Contract *bindings.KeyperSetManager
 	Log      log.Logger
 	Handler  event.KeyperSetHandler

rolling-shutter/medley/chainsync/syncer/shutterstate.go

@@ -16,7 +16,7 @@ import (
 var _ ManualFilterHandler = &ShutterStateSyncer{}
 
 type ShutterStateSyncer struct {
-	Client   client.EthereumClient
+	Client   client.SyncEthereumClient
 	Contract *bindings.KeyperSetManager
 	Log      log.Logger
 	Handler  event.ShutterStateHandler

rolling-shutter/medley/chainsync/syncer/unsafehead.go

@@ -17,7 +17,7 @@ import (
 )
 
 type UnsafeHeadSyncer struct {
-	Client  client.EthereumClient
+	Client  client.SyncEthereumClient
 	Log     log.Logger
 	Handler event.BlockHandler
 	// Handler to be manually triggered

rolling-shutter/medley/chainsync/syncer/unsafehead_test.go

@@ -0,0 +1,112 @@
+package syncer
+
+import (
+	"context"
+	"fmt"
+	"math/big"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/chainsync/client"
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/chainsync/event"
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/encodeable/number"
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/service"
+	"gotest.tools/v3/assert"
+)
+
+func MakeChain(start int64, startParent common.Hash, numHeader uint, seed int64) []*types.Header {
+	n := numHeader
+	parent := startParent
+	num := big.NewInt(start)
+	h := []*types.Header{}
+
+	// change the hashes for different seeds
+	mixinh := common.BigToHash(big.NewInt(seed))
+	for n > 0 {
+		head := &types.Header{
+			ParentHash: parent,
+			Number:     num,
+			MixDigest:  mixinh,
+		}
+		h = append(h, head)
+		num = new(big.Int).Add(num, big.NewInt(1))
+		parent = head.Hash()
+		n--
+	}
+	return h
+}
+
+func TestReorg(t *testing.T) {
+	headersBeforeReorg := MakeChain(1, common.BigToHash(big.NewInt(0)), 10, 42)
+	branchOff := headersBeforeReorg[5]
+	// block number 5 will be reorged
+	headersReorgBranch := MakeChain(branchOff.Number.Int64()+1, branchOff.Hash(), 10, 43)
+	clnt, ctl := client.NewTestClient()
+	ctl.AppendNextHeaders(headersBeforeReorg...)
+	ctl.AppendNextHeaders(headersReorgBranch...)
+
+	handlerBlock := make(chan *event.LatestBlock, 1)
+
+	h := &UnsafeHeadSyncer{
+		Client: clnt,
+		Log:    log.New(),
+		Handler: func(_ context.Context, ev *event.LatestBlock) error {
+			handlerBlock <- ev
+			return nil
+		},
+		SyncedHandler:      []ManualFilterHandler{},
+		SyncStartBlock:     number.NewBlockNumber(nil),
+		FetchActiveAtStart: false,
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+	defer cancel()
+	service.RunBackground(ctx, h)
+
+	// intitial sync is independent of the subscription,
+	// this will get polled from the eth client
+	b := <-handlerBlock
+	assert.Assert(t, b.Number.Cmp(headersBeforeReorg[0].Number) == 0)
+	idx := 1
+	for {
+		ok := ctl.ProgressHead()
+		assert.Assert(t, ok)
+		err := ctl.EmitEvents(ctx)
+		assert.NilError(t, err)
+
+		b = <-handlerBlock
+		assert.Equal(t, b.Number.Uint64(), headersBeforeReorg[idx].Number.Uint64(), fmt.Sprintf("block number equal for idx %d", idx))
+		assert.Equal(t, b.BlockHash, headersBeforeReorg[idx].Hash())
+		idx++
+		if idx == len(headersBeforeReorg) {
+			break
+		}
+	}
+	ok := ctl.ProgressHead()
+	assert.Assert(t, ok)
+	err := ctl.EmitEvents(ctx)
+	assert.NilError(t, err)
+	b = <-handlerBlock
+	// now the reorg should have happened.
+	// the handler should have emitted an "artificial" latest head
+	// event for the block BEFORE the re-orged block
+	assert.Equal(t, b.Number.Uint64(), headersReorgBranch[0].Number.Uint64()-1, "block number equal for reorg")
+	assert.Equal(t, b.BlockHash, headersReorgBranch[0].ParentHash)
+	idx = 0
+	for ctl.ProgressHead() {
+		assert.Assert(t, ok)
+		err := ctl.EmitEvents(ctx)
+		assert.NilError(t, err)
+
+		b := <-handlerBlock
+		assert.Equal(t, b.Number.Uint64(), headersReorgBranch[idx].Number.Uint64(), fmt.Sprintf("block number equal for idx %d", idx))
+		assert.Equal(t, b.BlockHash, headersReorgBranch[idx].Hash())
+		idx++
+		if idx == len(headersReorgBranch) {
+			break
+		}
+	}
+}