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,51 @@ 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()))
+
+			events, ok := sub.EventsBuffer.Buffer[newHead.Number]
+			if ok && events.BlockHash != newHead.Hash() {
+				// Deal with block reorg, delete all events with old hash value
+				sub.Logger.Info(fmt.Sprintf("Block reorg found old hash is %v new hash is %v", events.BlockHash, newHead.Hash()))
+				delete(sub.EventsBuffer.Buffer, newHead.Number)
+			}
+
+			// Iterate find out older enough events
+			for key, value := range sub.EventsBuffer.Buffer {
+				sub.Logger.Info(fmt.Sprintf("New header is %d stored header is %d", newHead.Number, key))
+				tmpKey := big.NewInt(30)
+				tmpKey.Add(tmpKey, key)
+				// event happened 30 blocks ago
+				if newHead.Number.Cmp(tmpKey) > 0 {
+					for _, event := range value.Events {
+						sub.Logger.Info(fmt.Sprintf("Process event %v", event))
+						err := sub.handleEthereumEvent(event)
+
+						if err != nil {
+							sub.Logger.Error(err.Error())
+							completionEvent.Add(1)
+						}
+					}
+					// delete the block after processing all events
+					delete(sub.EventsBuffer.Buffer, key)
+				}
+			}
+
 		// 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 +255,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 +292,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,62 @@
+package types
+
+import (
+	"errors"
+	"math/big"
+
+	"github.com/ethereum/go-ethereum/common"
+)
+
+// Events store all events in a block
+type Events struct {
+	BlockHash common.Hash
+	Events    []EthereumEvent
+}
+
+// AddEvents append event
+func (e Events) AddEvents(event EthereumEvent) (Events, error) {
+	for _, n := range e.Events {
+		if n.Equal(event) {
+			return Events{}, errors.New("event already in list")
+		}
+	}
+	return Events{
+		BlockHash: e.BlockHash,
+		Events:    append(e.Events, event),
+	}, nil
+}
+
+// EthEventBuffer store all events from Ethereum smart contract
+type EthEventBuffer struct {
+	Buffer map[*big.Int]Events
+}
+
+// NewEthEventBuffer create a new instance of EthEventBuffer
+func NewEthEventBuffer() EthEventBuffer {
+	return EthEventBuffer{
+		Buffer: make(map[*big.Int]Events),
+	}
+}
+
+// AddEvent insert a new event to queue
+func (buff EthEventBuffer) AddEvent(blockNumber *big.Int, blockHash common.Hash, event EthereumEvent) error {
+	events, ok := buff.Buffer[blockNumber]
+	if ok {
+		if blockHash == events.BlockHash {
+			newEvents, err := events.AddEvents(event)
+			if err != nil {
+				return err
+			}
+			buff.Buffer[blockNumber] = newEvents
+		} else {
+			// different hash with the same height
+			return errors.New("different event's block hash with the same block height")
+		}
+	} else {
+		buff.Buffer[blockNumber] = Events{
+			BlockHash: blockHash,
+			Events:    []EthereumEvent{event},
+		}
+	}
+	return nil
+}

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
+

test/integration/peggy-basic-demo.sh

@@ -3,48 +3,48 @@
 ## Case 1
 ## 1. send tx to cosmos after get the lock event in ethereum
 sifnodecli tx ethbridge create-claim 0x30753E4A8aad7F8597332E813735Def5dD395028 3 eth 0x11111111262b236c9ac9a9a8c8e4276b5cf6b2c9 \
-$(sifnodecli keys show user2 -a) $(sifnodecli keys show user1 -a --bech val) 5 lock \
---token-contract-address=0x0000000000000000000000000000000000000000 --ethereum-chain-id=3 --from=user1 --yes
+$(sifnodecli keys show akasha -a) $(sifnodecli keys show shadowfiend -a --bech val) 5 lock \
+--token-contract-address=0x0000000000000000000000000000000000000000 --ethereum-chain-id=3 --from=shadowfiend --yes
 
 # 2. query the tx
 #sifnodecli q tx
 
-# 3. check user2 account balance
-sifnodecli q auth account $(sifnodecli keys show user2 -a)
+# 3. check akasha account balance
+sifnodecli q auth account $(sifnodecli keys show akasha -a)
 
 # 4. query the prophecy
 sifnodecli query ethbridge prophecy 0x30753E4A8aad7F8597332E813735Def5dD395028 3 eth 0x11111111262b236c9ac9a9a8c8e4276b5cf6b2c9 --ethereum-chain-id=3 --token-contract-address=0x0000000000000000000000000000000000000000
 
 ## Case 2
-## 1. burn peggyetch for user2
-sifnodecli tx ethbridge burn $(sifnodecli keys show user2 -a) 0x11111111262b236c9ac9a9a8c8e4276b5cf6b2c9 \
-1 ceth --ethereum-chain-id=3 --from=user2 --yes
+## 1. burn peggyetch for akasha
+sifnodecli tx ethbridge burn $(sifnodecli keys show akasha -a) 0x11111111262b236c9ac9a9a8c8e4276b5cf6b2c9 \
+1 ceth --ethereum-chain-id=3 --from=akasha --yes
 
 ## 2. query the tx
 #sifnodecli q tx
 
-## 3. check user2 account balance
-sifnodecli q auth account $(sifnodecli keys show user2 -a)
+## 3. check akasha account balance
+sifnodecli q auth account $(sifnodecli keys show akasha -a)
 
 ## Case 3
-## 1. lock user2 rwn in sifchain
-sifnodecli tx ethbridge lock $(sifnodecli keys show user2 -a) 0x11111111262b236c9ac9a9a8c8e4276b5cf6b2c9 \
-10 rwn  --ethereum-chain-id=3 --from=user2 --yes
+## 1. lock akasha rwn in sifchain
+sifnodecli tx ethbridge lock $(sifnodecli keys show akasha -a) 0x11111111262b236c9ac9a9a8c8e4276b5cf6b2c9 \
+10 rwn  --ethereum-chain-id=3 --from=akasha --yes
 
 ## 2. query the tx
 #sifnodecli q tx
 
-## 3. check user2 account balance
-sifnodecli q auth account $(sifnodecli keys show user2 -a)
+## 3. check akasha account balance
+sifnodecli q auth account $(sifnodecli keys show akasha -a)
 
 ## Case 4
 ## 1. send tx to cosmos after peggyrwn burn in ethereum
 sifnodecli tx ethbridge create-claim 0x30753E4A8aad7F8597332E813735Def5dD395028 1 rwn 0x11111111262b236c9ac9a9a8c8e4276b5cf6b2c9 \
-$(sifnodecli keys show user2 -a) $(sifnodecli keys show user1 -a --bech val) \
-1 burn --ethereum-chain-id=3 --token-contract-address=0x345cA3e014Aaf5dcA488057592ee47305D9B3e10 --from=user1 --yes
+$(sifnodecli keys show akasha -a) $(sifnodecli keys show shadowfiend -a --bech val) \
+1 burn --ethereum-chain-id=3 --token-contract-address=0x345cA3e014Aaf5dcA488057592ee47305D9B3e10 --from=shadowfiend --yes
 
 ## 2. query the tx
 #sifnodecli q tx
 
-## 3. check user2 account balance
-sifnodecli q auth account $(sifnodecli keys show user2 -a)
+## 3. check akasha account balance
+sifnodecli q auth account $(sifnodecli keys show akasha -a)

test/integration/peggy-basic-test.py

@@ -3,8 +3,8 @@
 import time
 
 # define users
-VALIDATOR = "user1"
-USER = "user2"
+VALIDATOR = "shadowfiend"
+USER = "akasha"
 ROWAN = "rwn"
 PEGGYETH = "ceth"
 PEGGYROWAN = "erwn"