feature wait 50 blocks to process Ethereum events

cmd/ebrelayer/relayer/ethereum.go

@@ -50,9 +50,11 @@ type EthereumSub struct {
 	TxBldr                  authtypes.TxBuilder
 	PrivateKey              *ecdsa.PrivateKey
 	TempPassword            string
+	EventsBuffer            types.EthEventBuffer
 	Logger                  tmLog.Logger
 }
 
+// NewKeybase create a new keybase instance
 func NewKeybase(validatorMoniker, mnemonic, password string) (keys.Keybase, keys.Info, error) {
 	keybase := keys.NewInMemory()
 	hdpath := *hd.NewFundraiserParams(0, sdk.CoinType, 0)
@@ -97,6 +99,7 @@ func NewEthereumSub(inBuf io.Reader, rpcURL string, cdc *codec.Codec, validatorM
 		TxBldr:                  txBldr,
 		PrivateKey:              privateKey,
 		TempPassword:            tempPassword,
+		EventsBuffer:            types.NewEthEventBuffer(),
 		Logger:                  logger,
 	}, nil
 }
@@ -158,8 +161,16 @@ func (sub EthereumSub) Start(completionEvent *sync.WaitGroup) {
 	bridgeBankAddress, subBridgeBank := sub.startContractEventSub(logs, client, txs.BridgeBank)
 	defer subBridgeBank.Unsubscribe()
 	bridgeBankContractABI := contract.LoadABI(txs.BridgeBank)
-	eventLogLockSignature := bridgeBankContractABI.Events[types.LogLock.String()].ID().Hex()
-	eventLogBurnSignature := bridgeBankContractABI.Events[types.LogBurn.String()].ID().Hex()
+
+	// Listen the new header
+	heads := make(chan *ctypes.Header)
+	defer close(heads)
+	subHead, err := client.SubscribeNewHead(context.Background(), heads)
+	if err != nil {
+		log.Println(err)
+		return
+	}
+	defer subHead.Unsubscribe()
 
 	for {
 		select {
@@ -171,26 +182,47 @@ func (sub EthereumSub) Start(completionEvent *sync.WaitGroup) {
 			completionEvent.Add(1)
 			go sub.Start(completionEvent)
 			return
+		case err := <-subHead.Err():
+			sub.Logger.Error(err.Error())
+			completionEvent.Add(1)
+			go sub.Start(completionEvent)
+			return
+		case newHead := <-heads:
+			sub.Logger.Info(fmt.Sprintf("New header %d with hash %v", newHead.Number, newHead.Hash()))
+
+			// Add new header info to buffer
+			sub.EventsBuffer.AddHeader(newHead.Number, newHead.Hash(), newHead.ParentHash)
+
+			for {
+				thirty := big.NewInt(30)
+				thirty.Add(thirty, sub.EventsBuffer.MinHeight)
+				if thirty.Cmp(newHead.Number) <= 0 {
+					events := sub.EventsBuffer.GetHeaderEvents()
+					for _, event := range events {
+						err := sub.handleEthereumEvent(event)
+
+						if err != nil {
+							sub.Logger.Error(err.Error())
+							completionEvent.Add(1)
+						}
+					}
+
+					sub.EventsBuffer.RemoveHeight()
+
+				} else {
+					break
+				}
+			}
+
 		// vLog is raw event data
 		case vLog := <-logs:
 			sub.Logger.Info(fmt.Sprintf("Witnessed tx %s on block %d\n", vLog.TxHash.Hex(), vLog.BlockNumber))
-			log.Println("Found event from the ethereum bridgebank contract: ", types.LogLock.String())
-			var err error
-			switch vLog.Topics[0].Hex() {
-			case eventLogBurnSignature:
-				err = sub.handleEthereumEvent(clientChainID, bridgeBankAddress, bridgeBankContractABI,
-					types.LogBurn.String(), vLog)
-			case eventLogLockSignature:
-				log.Println("Found loglock event from the ethereum bridgebank contract: ", types.LogLock.String())
-				err = sub.handleEthereumEvent(clientChainID, bridgeBankAddress, bridgeBankContractABI,
-					types.LogLock.String(), vLog)
-			}
-			// TODO: Check local events store for status, if retryable, attempt relay again
+			event, isBurnLock, err := sub.logToEvent(clientChainID, bridgeBankAddress, bridgeBankContractABI, vLog)
 			if err != nil {
-				sub.Logger.Error(err.Error())
-				completionEvent.Add(1)
-				go sub.Start(completionEvent)
-				return
+				sub.Logger.Error("Failed to get event from ethereum log")
+			} else if isBurnLock {
+				sub.Logger.Info("Add event into buffer")
+				sub.EventsBuffer.AddEvent(big.NewInt(int64(vLog.BlockNumber)), vLog.BlockHash, event)
 			}
 		}
 	}
@@ -219,14 +251,31 @@ func (sub EthereumSub) startContractEventSub(logs chan ctypes.Log, client *ethcl
 	return subContractAddress, contractSub
 }
 
-// handleEthereumEvent unpacks an Ethereum event, converts it to a ProphecyClaim, and relays a tx to Cosmos
-func (sub EthereumSub) handleEthereumEvent(clientChainID *big.Int, contractAddress common.Address,
-	contractABI abi.ABI, eventName string, cLog ctypes.Log) error {
+// logToEvent unpacks an Ethereum event
+func (sub EthereumSub) logToEvent(clientChainID *big.Int, contractAddress common.Address,
+	contractABI abi.ABI, cLog ctypes.Log) (types.EthereumEvent, bool, error) {
 	// Parse the event's attributes via contract ABI
 	event := types.EthereumEvent{}
+	eventLogLockSignature := contractABI.Events[types.LogLock.String()].ID().Hex()
+	eventLogBurnSignature := contractABI.Events[types.LogBurn.String()].ID().Hex()
+
+	var eventName string
+	switch cLog.Topics[0].Hex() {
+	case eventLogBurnSignature:
+		eventName = types.LogBurn.String()
+	case eventLogLockSignature:
+		eventName = types.LogLock.String()
+	}
+
+	// If event is not expected
+	if eventName == "" {
+		return event, false, nil
+	}
+
 	err := contractABI.Unpack(&event, eventName, cLog.Data)
 	if err != nil {
-		sub.Logger.Error("error unpacking: %v", err)
+		sub.Logger.Error(err.Error())
+		return event, false, err
 	}
 	event.BridgeContractAddress = contractAddress
 	event.EthereumChainID = clientChainID
@@ -239,7 +288,11 @@ func (sub EthereumSub) handleEthereumEvent(clientChainID *big.Int, contractAddre
 
 	// Add the event to the record
 	types.NewEventWrite(cLog.TxHash.Hex(), event)
+	return event, true, nil
+}
 
+// handleEthereumEvent unpacks an Ethereum event, converts it to a ProphecyClaim, and relays a tx to Cosmos
+func (sub EthereumSub) handleEthereumEvent(event types.EthereumEvent) error {
 	prophecyClaim, err := txs.EthereumEventToEthBridgeClaim(sub.ValidatorAddress, &event)
 	if err != nil {
 		return err

cmd/ebrelayer/types/ethEventBuffer.go

@@ -0,0 +1,165 @@
+package types
+
+import (
+	"fmt"
+	"log"
+	"math/big"
+
+	"github.com/ethereum/go-ethereum/common"
+)
+
+// EventsInBlock store all events in a block, parent hash used to determine the best chain
+type EventsInBlock struct {
+	ParentHash common.Hash
+	Events     []EthereumEvent
+}
+
+// NewEventsInBlock create new instance with parent hash
+func NewEventsInBlock() EventsInBlock {
+	return EventsInBlock{
+		ParentHash: common.Hash{},
+		Events:     []EthereumEvent{},
+	}
+}
+
+// AddEvent append new event to list
+func (e *EventsInBlock) AddEvent(event EthereumEvent) {
+	// avoid add the same event twice
+	for _, n := range e.Events {
+		if n.Equal(event) {
+			log.Println("EventsInBlock event already in list")
+			return
+		}
+	}
+	e.Events = append(e.Events, event)
+}
+
+// EventsInHeight store all events at the same height
+type EventsInHeight struct {
+	// map the block hash to its parent hash and event list
+	EventsMap map[common.Hash]*EventsInBlock
+}
+
+// NewEventsInHeight create a new instance
+func NewEventsInHeight() EventsInHeight {
+	return EventsInHeight{
+		EventsMap: make(map[common.Hash]*EventsInBlock),
+	}
+}
+
+// AddEvent append event
+func (e *EventsInHeight) AddEvent(blockHash common.Hash, event EthereumEvent) {
+	events, ok := e.EventsMap[blockHash]
+	if ok {
+		events.AddEvent(event)
+	} else {
+		newEventsInBlock := NewEventsInBlock()
+		newEventsInBlock.AddEvent(event)
+		e.EventsMap[blockHash] = &newEventsInBlock
+	}
+}
+
+// AddHeader add a new block hash into map
+func (e *EventsInHeight) AddHeader(blockHash common.Hash, parentHash common.Hash) {
+	events, ok := e.EventsMap[blockHash]
+	// the events list the block hash already existed, then update the parent hash
+	if ok {
+		events.ParentHash = parentHash
+	} else {
+		newEventsInBlock := NewEventsInBlock()
+		newEventsInBlock.ParentHash = parentHash
+		e.EventsMap[blockHash] = &newEventsInBlock
+	}
+}
+
+// EthEventBuffer store all events from Ethereum smart contract
+type EthEventBuffer struct {
+	Buffer    map[string]EventsInHeight
+	MinHeight *big.Int
+}
+
+// NewEthEventBuffer create a new instance of EthEventBuffer
+func NewEthEventBuffer() EthEventBuffer {
+	return EthEventBuffer{
+		Buffer:    make(map[string]EventsInHeight),
+		MinHeight: big.NewInt(0),
+	}
+}
+
+// AddEvent insert a new event to queue
+// func (buff *EthEventBuffer) AddEvent(blockNumber *big.Int, blockHash common.Hash, event EthereumEvent) {
+func (buff *EthEventBuffer) AddEvent(blockNumber fmt.Stringer, blockHash common.Hash, event EthereumEvent) {
+	// Check if block number already in the map
+	events, ok := buff.Buffer[blockNumber.String()]
+	if ok {
+		events.AddEvent(blockHash, event)
+	} else {
+		newEvents := NewEventsInHeight()
+		newEvents.AddEvent(blockHash, event)
+		buff.Buffer[blockNumber.String()] = newEvents
+	}
+}
+
+// AddHeader create new entry for new header
+func (buff *EthEventBuffer) AddHeader(blockNumber *big.Int, blockHash common.Hash, parentHash common.Hash) {
+	if buff.MinHeight.Cmp(big.NewInt(0)) == 0 {
+		buff.MinHeight = blockNumber
+	}
+	// Check if block number already in the map
+	eventsInHeight, ok := buff.Buffer[blockNumber.String()]
+	if ok {
+		eventsInHeight.AddHeader(blockHash, parentHash)
+	} else {
+		newEventsInHeight := NewEventsInHeight()
+		newEventsInHeight.AddHeader(blockHash, parentHash)
+		buff.Buffer[blockNumber.String()] = newEventsInHeight
+	}
+}
+
+// GetDepth get the depth of a block
+func (buff *EthEventBuffer) GetDepth(blockNumber *big.Int, blockHash common.Hash) uint64 {
+	eventsInHeight, ok := buff.Buffer[blockNumber.String()]
+	if ok {
+		// if there is block's parent is the block hash
+		for key, eventsInBlock := range eventsInHeight.EventsMap {
+			if eventsInBlock.ParentHash == blockHash {
+				one := big.NewInt(1)
+				one.Add(one, blockNumber)
+
+				// recursive to its child block
+				return buff.GetDepth(one, key) + 1
+			}
+		}
+	}
+	return 0
+}
+
+// RemoveHeight remove an entry
+func (buff *EthEventBuffer) RemoveHeight() {
+	delete(buff.Buffer, buff.MinHeight.String())
+	buff.MinHeight.Add(buff.MinHeight, big.NewInt(1))
+}
+
+// GetHeaderEvents get the events in block of best chain
+func (buff *EthEventBuffer) GetHeaderEvents() []EthereumEvent {
+	eventsInHeight, ok := buff.Buffer[buff.MinHeight.String()]
+	if ok {
+		maxDepth := uint64(0)
+		result := []EthereumEvent{}
+		one := big.NewInt(1)
+		one.Add(one, buff.MinHeight)
+		for blockHash, eventsInBlock := range eventsInHeight.EventsMap {
+
+			depth := buff.GetDepth(one, blockHash)
+
+			if depth >= maxDepth {
+				maxDepth = depth
+				result = eventsInBlock.Events
+			}
+		}
+
+		return result
+	}
+
+	return []EthereumEvent{}
+}

cmd/ebrelayer/types/types.go

@@ -1,6 +1,7 @@
 package types
 
 import (
+	"bytes"
 	"fmt"
 	"math/big"
 
@@ -47,6 +48,19 @@ type EthereumEvent struct {
 	ClaimType             ethbridge.ClaimType
 }
 
+// Equal two events
+func (e EthereumEvent) Equal(other EthereumEvent) bool {
+	return e.EthereumChainID == other.EthereumChainID &&
+		e.BridgeContractAddress == other.BridgeContractAddress &&
+		bytes.Equal(e.ID[:], other.ID[:]) &&
+		e.From == other.From &&
+		bytes.Equal(e.To, other.To) &&
+		e.Symbol == other.Symbol &&
+		e.Value.Cmp(other.Value) == 0 &&
+		e.Nonce.Cmp(other.Nonce) == 0 &&
+		e.ClaimType == other.ClaimType
+}
+
 // String implements fmt.Stringer
 func (e EthereumEvent) String() string {
 	return fmt.Sprintf("\nChain ID: %v\nBridge contract address: %v\nToken symbol: %v\nToken "+

docs/adr-002-ebrelayer-eth-subscribe.md

@@ -0,0 +1,42 @@
+# ADR 002: Ebrelayer Ethereum Subscribe
+
+## Changelog
+
+- 2020/10/21: Initial version
+
+## Status
+
+*Proposed*
+
+## Context
+In this ADR, we discuss the solution for ebrelayer how to subscribe the events from Ethereum and process these events.
+### Summary
+
+For ebrelayer, it just needs to subscribe to the events from the BridgeBank smart contract, both LogLock event and LogBurn event, then process both events and send transaction to Sifchain. Basically, there are two problems.
+1. The block produced in Ethereum maybe be reorganized, most of system opt to confirm the finalization of block after 6 blocks. So it is better to process the event with some dalay. For the events once we receive from subscription, we store them in the buffer and wait for more blocks produced.
+2. How to store the events, there are two options. First one is store them in memory, but events will be lost if ebrelayer restarted. Second solution is store in local db, message queue like Kafka, but will increase the complexity of ebrelayer's deployment.
+
+## Current solution
+We start to process the events happened 30 blocks before. 30 blocks can guarantee the finalization of block. Then there is no impact from block reorganization. Repeated events and invalid events are abondoned in events buffer. We choose memory to store events for now, it is simple and easy to implements. The ebrelayer will miss some events if ebrelayer restarted. Considering the whole Sifchain system is a decentralized, we can tolerate some ebrelayer offline and not send transaction to Sifnode.  
+
+### Pros and Cons
+
+Pros: The solution is easy to implement. 
+
+Cons: The events lost if ebrelayer restart. We will store the events in persistent storage like local database or message queue system. It depends on the requirement of product in the future.
+
+## Consequences
+We will see obvious of delay for Sifchain get the message of cross-chain asset transfer. The end to end tests are impacted, need extra operations and transactions to verify the cases both transfering eth/erc20 asset to Sifchain and burn pegged Cosmos asset back to Sifchain.
+
+### Positive
+
+- We can quickly deliver our MVP
+
+### Negative
+
+- Nothing major
+
+### Neutral
+
+- Nothing major
+