feat(transaction): add transaction management

2020-07-17 10:29:36 +08:00 · 2020-07-17 10:29:36 +08:00 · e7b8d8b17c
parent 59ce87d0f3
commit e7b8d8b17c
8 changed files with 260 additions and 42 deletions
--- a/api/grpc/subscribe.go
+++ b/api/grpc/subscribe.go
@ -182,7 +182,9 @@ func (cbs *ChainBrokerService) interStatus(block *pb.Block, interchainMeta *pb.I
 			switch ibtp.Type {
 			case pb.IBTP_INTERCHAIN:
 				ev.InterchainTx = append(ev.InterchainTx, status)
-			case pb.IBTP_RECEIPT:
+			case pb.IBTP_RECEIPT_SUCCESS:
+				ev.InterchainReceipt = append(ev.InterchainReceipt, status)
+			case pb.IBTP_RECEIPT_FAILURE:
 				ev.InterchainReceipt = append(ev.InterchainReceipt, status)
 			}
 		}
--- a/go.mod
+++ b/go.mod
@ -23,7 +23,7 @@ require (
 	github.com/magiconair/properties v1.8.1
 	github.com/meshplus/bitxhub-core v0.1.0-rc1.0.20200526060151-b0efad4a2046
 	github.com/meshplus/bitxhub-kit v1.0.1-0.20200525060338-39d86f7542ae
-	github.com/meshplus/bitxhub-model v1.0.0-rc4.0.20200714081853-9d050b7250ad
+	github.com/meshplus/bitxhub-model v1.0.0-rc4.0.20200714082154-970b730cdb6c
 	github.com/mitchellh/go-homedir v1.1.0
 	github.com/multiformats/go-multiaddr v0.2.0
 	github.com/pkg/errors v0.9.1
--- a/go.sum
+++ b/go.sum
@ -512,6 +512,8 @@ github.com/meshplus/bitxhub-model v1.0.0-rc4.0.20200514093243-7e8ae60d1c19 h1:D0
 github.com/meshplus/bitxhub-model v1.0.0-rc4.0.20200514093243-7e8ae60d1c19/go.mod h1:QK8aACbxtZEA3Hk1BOCirW0uxMWLsMrLDpWz9FweIKM=
 github.com/meshplus/bitxhub-model v1.0.0-rc4.0.20200714081853-9d050b7250ad h1:lZQQlcwrI4BGKVdA4O2VW93K4P7wL04o3ODoFUESOBo=
 github.com/meshplus/bitxhub-model v1.0.0-rc4.0.20200714081853-9d050b7250ad/go.mod h1:QK8aACbxtZEA3Hk1BOCirW0uxMWLsMrLDpWz9FweIKM=
+github.com/meshplus/bitxhub-model v1.0.0-rc4.0.20200714082154-970b730cdb6c h1:RD+QMpgFKfdwy0rdmyaF2EX6Ris6pmz5scmGAuR/CYY=
+github.com/meshplus/bitxhub-model v1.0.0-rc4.0.20200714082154-970b730cdb6c/go.mod h1:QK8aACbxtZEA3Hk1BOCirW0uxMWLsMrLDpWz9FweIKM=
 github.com/mgutz/ansi v0.0.0-20170206155736-9520e82c474b/go.mod h1:01TrycV0kFyexm33Z7vhZRXopbI8J3TDReVlkTgMUxE=
 github.com/miekg/dns v1.1.12/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
 github.com/miekg/pkcs11 v1.0.3 h1:iMwmD7I5225wv84WxIG/bmxz9AXjWvTWIbM/TYHvWtw=
--- a/internal/constant/constant.go
+++ b/internal/constant/constant.go
@ -5,11 +5,12 @@ import "github.com/meshplus/bitxhub-kit/types"
 type BoltContractAddress string

 const (
-	InterchainContractAddr  BoltContractAddress = "0x000000000000000000000000000000000000000a"
-	StoreContractAddr       BoltContractAddress = "0x000000000000000000000000000000000000000b"
-	RuleManagerContractAddr BoltContractAddress = "0x000000000000000000000000000000000000000c"
-	RoleContractAddr        BoltContractAddress = "0x000000000000000000000000000000000000000d"
-	AppchainMgrContractAddr BoltContractAddress = "0x000000000000000000000000000000000000000e"
+	InterchainContractAddr     BoltContractAddress = "0x000000000000000000000000000000000000000a"
+	StoreContractAddr          BoltContractAddress = "0x000000000000000000000000000000000000000b"
+	RuleManagerContractAddr    BoltContractAddress = "0x000000000000000000000000000000000000000c"
+	RoleContractAddr           BoltContractAddress = "0x000000000000000000000000000000000000000d"
+	AppchainMgrContractAddr    BoltContractAddress = "0x000000000000000000000000000000000000000e"
+	TransactionMgrContractAddr BoltContractAddress = "0x000000000000000000000000000000000000000f"
 )

 func (addr BoltContractAddress) Address() types.Address {
--- a/internal/executor/contracts/interchain.go
+++ b/internal/executor/contracts/interchain.go
@ -88,82 +88,167 @@ func (x *InterchainManager) HandleIBTP(data []byte) *boltvm.Response {
 		return boltvm.Error(err.Error())
 	}

-	if ibtp.To == "" {
-		return boltvm.Error("empty destination chain id")
-	}
-	ok = x.Has(x.appchainKey(ibtp.To))
-	if !ok {
-		x.Logger().WithField("chain_id", ibtp.To).Warn("target appchain does not exist")
-	}
-
 	interchain := &Interchain{}
 	x.GetObject(x.appchainKey(ibtp.From), &interchain)

-	app := &appchainMgr.Appchain{}
-	res := x.CrossInvoke(constant.AppchainMgrContractAddr.String(), "GetAppchain", pb.String(ibtp.From))
-	err := json.Unmarshal(res.Result, app)
-	if err != nil {
+	if err := x.checkIBTP(ibtp, interchain); err != nil {
 		return boltvm.Error(err.Error())
 	}
+
+	res := boltvm.Success(nil)
+
+	if pb.IBTP_INTERCHAIN == ibtp.Type {
+		res = x.beginTransaction(ibtp)
+	} else {
+		res = x.reportTransaction(ibtp)
+	}
+
+	if !res.Ok {
+		return res
+	}
+
+	x.ProcessIBTP(ibtp, interchain)
+
+	return res
+}
+
+func (x *InterchainManager) HandleIBTPs(data []byte) *boltvm.Response {
+	ok := x.Has(x.appchainKey(x.Caller()))
+	if !ok {
+		return boltvm.Error("this appchain does not exist")
+	}
+
+	ibtps := &pb.IBTPs{}
+	if err := ibtps.Unmarshal(data); err != nil {
+		return boltvm.Error(err.Error())
+	}
+
+	interchain := &Interchain{}
+	x.GetObject(x.appchainKey(x.Caller()), &interchain)
+
+	for _, ibtp := range ibtps.Iptp {
+		if err := x.checkIBTP(ibtp, interchain); err != nil {
+			return boltvm.Error(err.Error())
+		}
+	}
+
+	if res := x.beginMultiTargetsTransaction(ibtps); !res.Ok {
+		return res
+	}
+
+	for _, ibtp := range ibtps.Iptp {
+		x.ProcessIBTP(ibtp, interchain)
+	}
+
+	return boltvm.Success(nil)
+}
+
+func (x *InterchainManager) checkIBTP(ibtp *pb.IBTP, interchain *Interchain) error {
+	if ibtp.To == "" {
+		return fmt.Errorf("empty destination chain id")
+	}
+	if ok := x.Has(x.appchainKey(ibtp.To)); !ok {
+		x.Logger().WithField("chain_id", ibtp.To).Warn("target appchain does not exist")
+	}
+
+	app := &appchainMgr.Appchain{}
+	res := x.CrossInvoke(constant.AppchainMgrContractAddr.String(), "GetAppchain", pb.String(ibtp.From))
+	if err := json.Unmarshal(res.Result, app); err != nil {
+		return err
+	}
+
 	// get validation rule contract address
 	res = x.CrossInvoke(constant.RuleManagerContractAddr.String(), "GetRuleAddress", pb.String(ibtp.From), pb.String(app.ChainType))
 	if !res.Ok {
-		return boltvm.Error("this appchain don't register rule")
+		return fmt.Errorf("this appchain does not register rule")
 	}

 	// handle validation
 	isValid, err := x.ValidationEngine().Validate(string(res.Result), ibtp.From, ibtp.Proof, ibtp.Payload, app.Validators)
 	if err != nil {
-		return boltvm.Error(err.Error())
+		return err
 	}

 	if !isValid {
-		return boltvm.Error("invalid interchain transaction")
+		return fmt.Errorf("invalid interchain transaction")
 	}

-	switch ibtp.Type {
-	case pb.IBTP_INTERCHAIN:
+	if pb.IBTP_INTERCHAIN == ibtp.Type {
 		if ibtp.From != x.Caller() {
-			return boltvm.Error("ibtp from != caller")
+			return fmt.Errorf("ibtp from != caller")
 		}

 		idx := interchain.InterchainCounter[ibtp.To]
 		if idx+1 != ibtp.Index {
-			return boltvm.Error(fmt.Sprintf("wrong index, required %d, but %d", idx+1, ibtp.Index))
+			return fmt.Errorf(fmt.Sprintf("wrong index, required %d, but %d", idx+1, ibtp.Index))
 		}
-
-		interchain.InterchainCounter[ibtp.To]++
-		x.SetObject(x.appchainKey(ibtp.From), interchain)
-		x.SetObject(x.indexMapKey(ibtp.ID()), x.GetTxHash())
-		m := make(map[string]uint64)
-		m[ibtp.To] = x.GetTxIndex()
-		x.PostInterchainEvent(m)
-	case pb.IBTP_RECEIPT:
+	} else {
 		if ibtp.To != x.Caller() {
-			return boltvm.Error("ibtp from != caller")
+			return fmt.Errorf("ibtp from != caller")
 		}

 		idx := interchain.ReceiptCounter[ibtp.To]
 		if idx+1 != ibtp.Index {
 			if interchain.SourceReceiptCounter[ibtp.To]+1 != ibtp.Index {
-				return boltvm.Error(fmt.Sprintf("wrong receipt index, required %d, but %d", idx+1, ibtp.Index))
+				return fmt.Errorf("wrong receipt index, required %d, but %d", idx+1, ibtp.Index)
 			}
 		}
+	}

+	return nil
+}
+
+func (x *InterchainManager) ProcessIBTP(ibtp *pb.IBTP, interchain *Interchain) {
+	m := make(map[string]uint64)
+
+	if pb.IBTP_INTERCHAIN == ibtp.Type {
+		interchain.InterchainCounter[ibtp.To]++
+		x.SetObject(x.appchainKey(ibtp.From), interchain)
+		x.SetObject(x.indexMapKey(ibtp.ID()), x.GetTxHash())
+		m[ibtp.To] = x.GetTxIndex()
+	} else {
 		interchain.ReceiptCounter[ibtp.To] = ibtp.Index
 		x.SetObject(x.appchainKey(ibtp.From), interchain)
-		m := make(map[string]uint64)
 		m[ibtp.From] = x.GetTxIndex()

 		ic := &Interchain{}
 		x.GetObject(x.appchainKey(ibtp.To), &ic)
 		ic.SourceReceiptCounter[ibtp.From] = ibtp.Index
 		x.SetObject(x.appchainKey(ibtp.To), ic)
-
-		x.PostInterchainEvent(m)
 	}

-	return boltvm.Success(nil)
+	x.PostInterchainEvent(m)
+}
+
+func (x *InterchainManager) beginMultiTargetsTransaction(ibtps *pb.IBTPs) *boltvm.Response {
+	args := make([]*pb.Arg, 0)
+	globalId := fmt.Sprintf("%s-%s", x.Caller(), x.GetTxHash())
+	args = append(args, pb.String(globalId))
+
+	for _, ibtp := range ibtps.Iptp {
+		if ibtp.Type != pb.IBTP_INTERCHAIN {
+			return boltvm.Error("ibtp type != IBTP_INTERCHAIN")
+		}
+
+		childTxId := fmt.Sprintf("%s-%s-%d", ibtp.From, ibtp.To, ibtp.Index)
+		args = append(args, pb.String(childTxId))
+	}
+
+	return x.CrossInvoke(constant.TransactionMgrContractAddr.String(), "Begin", args...)
+}
+
+func (x *InterchainManager) beginTransaction(ibtp *pb.IBTP) *boltvm.Response {
+	txId := fmt.Sprintf("%s-%s-%d", ibtp.From, ibtp.To, ibtp.Index)
+	return x.CrossInvoke(constant.TransactionMgrContractAddr.String(), "Begin", pb.String(txId))
+}
+
+func (x *InterchainManager) reportTransaction(ibtp *pb.IBTP) *boltvm.Response {
+	txId := fmt.Sprintf("%s-%s-%d", ibtp.To, ibtp.From, ibtp.Index)
+	result := int32(0)
+	if ibtp.Type == pb.IBTP_RECEIPT_FAILURE {
+		result = 1
+	}
+	return x.CrossInvoke(constant.TransactionMgrContractAddr.String(), "Report", pb.String(txId), pb.Int32(result))
 }

 func (x *InterchainManager) GetIBTPByID(id string) *boltvm.Response {
--- a/internal/executor/contracts/transaction_manager.go
+++ b/internal/executor/contracts/transaction_manager.go
@ -0,0 +1,121 @@
+package contracts
+
+import (
+	"fmt"
+
+	"github.com/meshplus/bitxhub/pkg/vm/boltvm"
+)
+
+type TransactionStatus string
+
+const (
+	PREFIX                          = "tx-"
+	StatusBegin   TransactionStatus = "BEGIN"
+	StatusSuccess TransactionStatus = "SUCCESS"
+	StatusFail    TransactionStatus = "FAIL"
+)
+
+type TransactionManager struct {
+	boltvm.Stub
+}
+
+type TransactionInfo struct {
+	globalState TransactionStatus
+	childTxInfo map[string]TransactionStatus
+}
+
+func (t *TransactionManager) BeginMultiTXs(globalId string, childTxIds ...string) *boltvm.Response {
+	if t.Has(t.txInfoKey(globalId)) {
+		return boltvm.Error("Transaction id already exists")
+	}
+
+	txInfo := &TransactionInfo{
+		globalState: StatusBegin,
+		childTxInfo: make(map[string]TransactionStatus),
+	}
+
+	for _, childTxId := range childTxIds {
+		txInfo.childTxInfo[childTxId] = StatusBegin
+		t.Set(childTxId, []byte(globalId))
+	}
+
+	t.SetObject(t.txInfoKey(globalId), txInfo)
+
+	return boltvm.Success(nil)
+}
+
+func (t *TransactionManager) Begin(txId string) *boltvm.Response {
+	if t.Has(t.txInfoKey(txId)) {
+		return boltvm.Error("Transaction id already exists")
+	}
+
+	t.Set(t.txInfoKey(txId), []byte(StatusBegin))
+
+	return boltvm.Success(nil)
+}
+
+func (t *TransactionManager) Report(txId string, result int32) *boltvm.Response {
+	ok, val := t.Get(t.txInfoKey(txId))
+	if ok {
+		status := TransactionStatus(val)
+		if status != StatusBegin {
+			return boltvm.Error(fmt.Sprintf("transaction with Id %s is finished", txId))
+		}
+
+		if result == 0 {
+			t.Set(t.txInfoKey(txId), []byte(StatusSuccess))
+		} else {
+			t.Set(t.txInfoKey(txId), []byte(StatusFail))
+		}
+	} else {
+		ok, val = t.Get(txId)
+		if !ok {
+			return boltvm.Error(fmt.Sprintf("cannot get global id of child tx id %s", txId))
+		}
+
+		globalId := string(val)
+		txInfo := &TransactionInfo{}
+		if !t.GetObject(t.txInfoKey(globalId), &txInfo) {
+			return boltvm.Error(fmt.Sprintf("transaction global id %s does not exist", globalId))
+		}
+
+		if txInfo.globalState != StatusBegin {
+			return boltvm.Error(fmt.Sprintf("transaction with global Id %s is finished", globalId))
+		}
+
+		status, ok := txInfo.childTxInfo[txId]
+		if !ok {
+			return boltvm.Error(fmt.Sprintf("%s is not in transaction %s", txId, globalId))
+		}
+
+		if status != StatusBegin {
+			return boltvm.Error(fmt.Sprintf("%s has already reported result", txId))
+		}
+
+		if result == 0 {
+			txInfo.childTxInfo[txId] = StatusSuccess
+			count := 0
+			for _, res := range txInfo.childTxInfo {
+				if res != StatusSuccess {
+					break
+				}
+				count++
+			}
+
+			if count == len(txInfo.childTxInfo) {
+				txInfo.globalState = StatusSuccess
+			}
+		} else {
+			txInfo.childTxInfo[txId] = StatusFail
+			txInfo.globalState = StatusFail
+		}
+
+		t.SetObject(t.txInfoKey(globalId), txInfo)
+	}
+
+	return boltvm.Success(nil)
+}
+
+func (t *TransactionManager) txInfoKey(id string) string {
+	return fmt.Sprintf("%s-%s", PREFIX, id)
+}
--- a/internal/executor/executor.go
+++ b/internal/executor/executor.go
@ -197,6 +197,12 @@ func registerBoltContracts() map[string]boltvm.Contract {
 			Address:  constant.AppchainMgrContractAddr.String(),
 			Contract: &contracts.AppchainManager{},
 		},
+		{
+			Enabled:  true,
+			Name:     "transaction manager service",
+			Address:  constant.TransactionMgrContractAddr.String(),
+			Contract: &contracts.TransactionManager{},
+		},
 	}

 	return boltvm.Register(boltContracts)
--- a/tester/case003_interchain_test.go
+++ b/tester/case003_interchain_test.go
@ -137,13 +137,14 @@ func (suite *Interchain) TestGetIBTPByID() {
 	ib := &pb.IBTP{From: f.Hex(), To: t.Hex(), Index: 1, Payload: []byte("111"), Timestamp: time.Now().UnixNano(), Proof: proof}
 	data, err := ib.Marshal()
 	suite.Require().Nil(err)
-	_, err = invokeBVMContract(suite.api, k1, constant.InterchainContractAddr.Address(), "HandleIBTP", pb.Bytes(data))
+	receipt, err := invokeBVMContract(suite.api, k1, constant.InterchainContractAddr.Address(), "HandleIBTP", pb.Bytes(data))
 	suite.Require().Nil(err)
+	suite.Require().EqualValues(true, receipt.IsSuccess(), string(receipt.Ret))

 	ib.Index = 2
 	data, err = ib.Marshal()
 	suite.Require().Nil(err)
-	receipt, err := invokeBVMContract(suite.api, k1, constant.InterchainContractAddr.Address(), "HandleIBTP", pb.Bytes(data))
+	receipt, err = invokeBVMContract(suite.api, k1, constant.InterchainContractAddr.Address(), "HandleIBTP", pb.Bytes(data))
 	suite.Require().Nil(err)
 	suite.Require().EqualValues(true, receipt.IsSuccess(), string(receipt.Ret))