Merge pull request #10 from icon-project/store-last-tx-info-to-storage

Store last TX info to storage

Author: eunsoo-icon

chain/eth2/eth2factory.go

@@ -40,14 +40,14 @@ func NewReceiver(srcCfg link.ChainConfig, dstAddr types.BtpAddress, baseDir stri
 	return newReceiver(srcCfg.GetAddress(), dstAddr, src.Endpoint, src.Options, l), nil
 }
 
-func NewSender(srcAddr types.BtpAddress, dstCfg link.ChainConfig, l log.Logger) (types.Sender, error) {
+func NewSender(srcAddr types.BtpAddress, dstCfg link.ChainConfig, baseDir string, l log.Logger) (types.Sender, error) {
 	dst := dstCfg.(chain.BaseConfig)
 	w, err := newWallet(dst.KeyStorePass, dst.KeySecret, dst.KeyStore)
 	if err != nil {
 		return nil, err
 	}
 
-	return newSender(srcAddr, dst.Address, w, dst.Endpoint, dst.Options, l), nil
+	return newSender(srcAddr, dst.Address, w, dst.Endpoint, dst.Options, baseDir, l), nil
 }
 
 func newWallet(passwd, secret string, keyStorePath string) (types.Wallet, error) {

chain/eth2/receiver_test.go

@@ -22,8 +22,8 @@ import (
 	"github.com/icon-project/btp2-eth2/chain/eth2/proof"
 )
 
-func newReceiver(src, dest types.BtpAddress) *receiver {
-	r := NewReceiver(
+func newTestReceiver(src, dest types.BtpAddress) *receiver {
+	r := newReceiver(
 		src,
 		dest,
 		"https://sepolia.infura.io/v3/ffbf8ebe228f4758ae82e175640275e0",
@@ -36,7 +36,7 @@ func newReceiver(src, dest types.BtpAddress) *receiver {
 }
 
 func TestReceiver_BlockUpdate(t *testing.T) {
-	r := newReceiver(
+	r := newTestReceiver(
 		types.BtpAddress("btp://0xaa36a7.eth/0x11167e875E08a113706e8bA3010ac37329b0E6b2"),
 		types.BtpAddress("btp://0x42.icon/cx8642ab29e608915b43e677d9bcb17ec902b4ec8b"),
 	)
@@ -113,7 +113,7 @@ func VerifySyncAggregate(t *testing.T, r *receiver, bu *blockUpdateData) {
 }
 
 func TestReceiver_BlockProof(t *testing.T) {
-	r := newReceiver(
+	r := newTestReceiver(
 		types.BtpAddress("btp://0xaa36a7.eth/0x11167e875E08a113706e8bA3010ac37329b0E6b2"),
 		types.BtpAddress("btp://0x42.icon/cx8642ab29e608915b43e677d9bcb17ec902b4ec8b"),
 	)
@@ -216,7 +216,7 @@ func TestReceiver_BlockProof(t *testing.T) {
 
 func TestReceiver_MessageProof(t *testing.T) {
 	slot := int64(2091171)
-	r := newReceiver(
+	r := newTestReceiver(
 		types.BtpAddress("btp://0xaa36a7.eth/0x11167e875E08a113706e8bA3010ac37329b0E6b2"),
 		types.BtpAddress("btp://0x42.icon/cx8642ab29e608915b43e677d9bcb17ec902b4ec8b"),
 	)

chain/eth2/sender.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"math"
 	"math/big"
+	"path/filepath"
 	"strconv"
 	"strings"
 	"sync"
@@ -30,63 +31,56 @@ import (
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/common"
 	etypes "github.com/ethereum/go-ethereum/core/types"
+	"github.com/icon-project/btp2/common/codec"
 	"github.com/icon-project/btp2/common/errors"
 	"github.com/icon-project/btp2/common/log"
 	"github.com/icon-project/btp2/common/types"
 	"github.com/icon-project/btp2/common/wallet"
+	"github.com/syndtr/goleveldb/leveldb"
 
 	"github.com/icon-project/btp2-eth2/chain/eth2/client"
 )
 
 const (
-	txMaxDataSize       = 524288 //512 * 1024 // 512kB
-	txOverheadScale     = 0.37   //base64 encoding overhead 0.36, rlp and other fields 0.01
-	GetTXResultInterval = SecondPerSlot * time.Second
-	txPendingMAX        = 3 // unit: finality
+	txMaxDataSize    = 524288 //512 * 1024 // 512kB
+	txOverheadScale  = 0.37   //base64 encoding overhead 0.36, rlp and other fields 0.01
+	blockInterval    = 12 * time.Second
+	txExpireDuration = 5 * blockInterval
 )
 
 var (
 	txSizeLimit = int(math.Ceil(txMaxDataSize / (1 + txOverheadScale)))
+	lastTXKey   = []byte("last_tx")
 )
 
 type request struct {
-	rm             types.RelayMessage
-	txHash         common.Hash
-	txPendingCount int
-}
-
-func (r *request) RelayMessage() types.RelayMessage {
-	return r.rm
+	rm types.RelayMessage
+	transaction
 }
 
 func (r *request) ID() string {
 	return r.rm.Id()
 }
 
-func (r *request) TxHash() common.Hash {
-	return r.txHash
-}
-
-func (r *request) SetTxHash(txHash common.Hash) {
-	r.txHash = txHash
-}
-
-func (r *request) SetTxPendingCount(value int) {
-	r.txPendingCount = value
-}
-
-func (r *request) IncTxPendingCount() int {
-	r.txPendingCount += 1
-	return r.txPendingCount
+func newRequest(rm types.RelayMessage, txHash common.Hash) *request {
+	now := time.Now()
+	return &request{
+		rm: rm,
+		transaction: transaction{
+			TxHash:   txHash,
+			TS:       now,
+			ExpireTS: now.Add(txExpireDuration),
+		},
+	}
 }
 
 func (r *request) Format(f fmt.State, c rune) {
 	switch c {
 	case 'v', 's':
 		if f.Flag('+') {
-			fmt.Fprintf(f, "request{id=%s txHash=%#x txPendingCount=%d)", r.ID(), r.txHash, r.txPendingCount)
+			fmt.Fprintf(f, "request{id=%s tx=%+v)", r.ID(), r.transaction)
 		} else {
-			fmt.Fprintf(f, "request{%s %#x %d)", r.ID(), r.txHash, r.txPendingCount)
+			fmt.Fprintf(f, "request{%s %v)", r.ID(), r.transaction)
 		}
 	}
 }
@@ -99,6 +93,7 @@ type sender struct {
 	sc   chan *types.RelayResult
 	reqs []*request
 	mtx  sync.RWMutex
+	db   *leveldb.DB
 
 	cl  *client.ConsensusLayer
 	el  *client.ExecutionLayer
@@ -107,7 +102,7 @@ type sender struct {
 	gasLimit uint64
 }
 
-func newSender(src, dst types.BtpAddress, w types.Wallet, endpoint string, opt map[string]interface{}, l log.Logger) types.Sender {
+func newSender(src, dst types.BtpAddress, w types.Wallet, endpoint string, opt map[string]interface{}, baseDir string, l log.Logger) types.Sender {
 	var err error
 	s := &sender{
 		src: src,
@@ -116,6 +111,11 @@ func newSender(src, dst types.BtpAddress, w types.Wallet, endpoint string, opt m
 		l:   l,
 		sc:  make(chan *types.RelayResult),
 	}
+
+	if err = s.initDatabase(baseDir); err != nil {
+		l.Panicf("fail to initialize database %v", err)
+	}
+
 	s.el, err = client.NewExecutionLayer(endpoint, l)
 	if err != nil {
 		l.Panicf("fail to connect to %s, %v", endpoint, err)
@@ -137,6 +137,24 @@ func newSender(src, dst types.BtpAddress, w types.Wallet, endpoint string, opt m
 	return s
 }
 
+func (s *sender) initDatabase(baseDir string) error {
+	var err error
+	//dbDir := filepath.Join(baseDir, s.dst.NetworkAddress(), s.src.NetworkAddress())
+	dbDir := filepath.Join(baseDir, s.src.NetworkAddress(), s.src.ContractAddress(),
+		"sender", s.dst.NetworkAddress(), s.dst.ContractAddress())
+	s.l.Debugln("open database", dbDir)
+	s.db, err = leveldb.OpenFile(dbDir, nil)
+	if err != nil {
+		return errors.Wrap(err, "fail to open database")
+	}
+	defer func() {
+		if err != nil {
+			s.db.Close()
+		}
+	}()
+	return nil
+}
+
 func (s *sender) Start() (<-chan *types.RelayResult, error) {
 	go s.handleFinalityUpdate()
 
@@ -155,7 +173,12 @@ func (s *sender) Relay(rm types.RelayMessage) (string, error) {
 	if tx, err := s.relay(rm); err != nil {
 		return rm.Id(), err
 	} else {
-		s.addRequest(&request{rm: rm, txHash: tx.Hash()})
+		req := newRequest(rm, tx.Hash())
+		s.addRequest(req)
+		if err = s.putLastTransaction(&req.transaction); err != nil {
+			s.removeRequest(rm.Id())
+			return rm.Id(), err
+		}
 	}
 	return rm.Id(), nil
 }
@@ -201,6 +224,23 @@ func (s *sender) removeRequest(id string) {
 }
 
 func (s *sender) getStatus(bn uint64) (*types.BMCLinkStatus, error) {
+	if ltx, err := s.getLastTransaction(); err != nil {
+		s.l.Errorf("fail to get last TX. %v", err)
+		return nil, err
+	} else if ltx != nil {
+		s.l.Debugf("last TX. %+v", ltx)
+		var pending bool
+		if _, pending, err = s.el.TransactionByHash(ltx.TxHash); err != nil {
+			s.l.Errorf("fail to query last TX %#x. %v", ltx.TxHash, err)
+			return nil, err
+		} else if pending {
+			if !ltx.IsExpired() {
+				s.l.Debugf("last TX %#x is not yet executed. sleep to %s", ltx.TxHash, ltx.ExpireTS)
+				time.Sleep(ltx.GetDurationToExpire())
+			}
+		}
+	}
+
 	var status client.TypesLinkStatus
 	var callOpts *bind.CallOpts
 	if bn != 0 {
@@ -238,43 +278,41 @@ func (s *sender) handleFinalityUpdate() {
 	}
 }
 
+func (s *sender) sendRelayResult(id string, errCode errors.Code, finalized bool) {
+	s.sc <- &types.RelayResult{Id: id, Err: errCode, Finalized: finalized}
+}
+
 func (s *sender) checkRelayResult(to uint64) {
 	finished := make([]*request, 0)
 	s.mtx.RLock()
-	for i, req := range s.reqs {
-		_, pending, err := s.el.TransactionByHash(req.TxHash())
+	for _, req := range s.reqs {
+		_, pending, err := s.el.TransactionByHash(req.TxHash)
 		if err != nil {
-			s.l.Warnf("can't get TX %#x. %v", req.TxHash(), err)
+			s.l.Errorf("can't get TX via rpc. %+v. Drop %+v.", err, req)
+			s.sendRelayResult(req.ID(), errors.BMVUnknown, false)
 			break
 		}
 		if pending {
-			s.l.Debugf("TX %#x is not yet executed.", req.TxHash())
-			if req.IncTxPendingCount() == txPendingMAX {
-				s.l.Debugf("resend rm %s", req.ID())
-				if tx, err := s.relay(req.RelayMessage()); err != nil {
-					s.l.Errorf("fail to resend relay message %s", req.ID())
-				} else {
-					req.SetTxHash(tx.Hash())
-					req.SetTxPendingCount(0)
-				}
+			if req.IsExpired() {
+				s.l.Errorf("request is still pending. Drop %+v.", req)
+				s.sendRelayResult(req.ID(), errors.BMVUnknown, false)
+				break
 			}
-			s.reqs[i] = req
-			s.l.Debugf("update req: %+v", s.reqs[i])
-			break
 		}
-		receipt, err := s.el.TransactionReceipt(req.TxHash())
+		receipt, err := s.el.TransactionReceipt(req.TxHash)
 		if err != nil {
-			s.l.Warnf("can't get TX receipt for %#x. %v", req.TxHash(), err)
+			s.l.Errorf("can't get TX receipt. %+v. Drop %+v.", err, req)
+			s.sendRelayResult(req.ID(), errors.BMVUnknown, false)
 			break
 		}
 		if to < receipt.BlockNumber.Uint64() {
-			s.l.Debugf("%#x is not yet finalized", req.TxHash())
+			s.l.Debugf("%#x is not yet finalized", req.TxHash)
 			break
 		}
 		err = s.receiptToRevertError(receipt)
 		errCode := errors.SUCCESS
 		if err != nil {
-			s.l.Debugf("result fail %v. %v", req, err)
+			s.l.Debugf("result fail %+v. %v", req, err)
 			if ec, ok := errors.CoderOf(err); ok {
 				errCode = ec.ErrorCode()
 			} else {
@@ -283,11 +321,7 @@ func (s *sender) checkRelayResult(to uint64) {
 		} else {
 			s.l.Debugf("result success. %v", req)
 		}
-		s.sc <- &types.RelayResult{
-			Id:        req.ID(),
-			Err:       errCode,
-			Finalized: true,
-		}
+		s.sendRelayResult(req.ID(), errCode, true)
 		finished = append(finished, req)
 	}
 	s.mtx.RUnlock()
@@ -316,3 +350,85 @@ func (s *sender) receiptToRevertError(receipt *etypes.Receipt) error {
 	}
 	return nil
 }
+
+func (s *sender) putLastTransaction(tx *transaction) error {
+	if bs, err := tx.Bytes(); err != nil {
+		return err
+	} else {
+		if err = s.db.Put(lastTXKey, bs, nil); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (s *sender) getLastTransaction() (*transaction, error) {
+	var bs []byte
+	var has bool
+	var err error
+	if has, err = s.db.Has(lastTXKey, nil); err != nil {
+		return nil, err
+	} else if has {
+		if bs, err = s.db.Get(lastTXKey, nil); err != nil {
+			return nil, err
+		} else {
+			return newTransactionFromBytes(bs)
+		}
+	} else {
+		return nil, nil
+	}
+}
+
+type transaction struct {
+	TxHash   common.Hash
+	TS       time.Time
+	ExpireTS time.Time // wait for TX propagation and pending time
+}
+
+func (tx *transaction) IsExpired() bool {
+	return time.Now().After(tx.ExpireTS)
+}
+
+func (tx *transaction) GetDurationToExpire() time.Duration {
+	ts := tx.ExpireTS
+	return ts.Sub(time.Now())
+}
+
+func (tx *transaction) Bytes() ([]byte, error) {
+	var bytes []byte
+	if bs, err := codec.MarshalToBytes(&tx); err != nil {
+		return nil, err
+	} else {
+		bytes = bs
+	}
+	return bytes, nil
+}
+
+func (tx *transaction) Format(f fmt.State, c rune) {
+	switch c {
+	case 'v', 's':
+		if f.Flag('+') {
+			fmt.Fprintf(f, "transaction{txHash=%#x TS=%s expireTS=%s)",
+				tx.TxHash, tx.TS.Format(time.RFC3339), tx.ExpireTS.Format(time.RFC3339))
+		} else {
+			fmt.Fprintf(f, "transaction{%#x %s %s)",
+				tx.TxHash, tx.TS.Format(time.RFC3339), tx.ExpireTS.Format(time.RFC3339))
+		}
+	}
+}
+
+func newTransaction(ts time.Time, txHash common.Hash) *transaction {
+	return &transaction{
+		TxHash:   txHash,
+		TS:       ts,
+		ExpireTS: ts.Add(txExpireDuration),
+	}
+}
+
+func newTransactionFromBytes(bs []byte) (*transaction, error) {
+	ltx := new(transaction)
+	if _, err := codec.UnmarshalFromBytes(bs, &ltx); err != nil {
+		return nil, err
+	}
+	return ltx, nil
+}