paracross self consensus add fork

plugin/consensus/para/para.go

@@ -47,9 +47,11 @@ var (
 	emptyBlockInterval int64 = 4 //write empty block every interval blocks in mainchain
 	zeroHash           [32]byte
 	//current miner tx take any privatekey for unify all nodes sign purpose, and para chain is free
-	minerPrivateKey               = "6da92a632ab7deb67d38c0f6560bcfed28167998f6496db64c258d5e8393a81b"
-	searchHashMatchDepth    int32 = 100
-	mainBlockHashForkHeight int64 = types.MaxHeight //calc block hash fork height in main chain
+	minerPrivateKey                  = "6da92a632ab7deb67d38c0f6560bcfed28167998f6496db64c258d5e8393a81b"
+	searchHashMatchDepth       int32 = 100
+	mainBlockHashForkHeight    int64 = types.MaxHeight //calc block hash fork height in main chain
+	mainParaCommitTxForkHeight int64 = types.MaxHeight //support paracross commit tx fork height in main chain
+	curMainChainHeight         int64
 )
 
 func init() {
@@ -173,11 +175,6 @@ func (client *client) SetQueueClient(c queue.Client) {
 
 func (client *client) InitBlock() {
 	var err error
-	// get main chain calc block hash fork height
-	mainBlockHashForkHeight, err = client.GetBlockHashForkHeightOnMainChain()
-	if err != nil {
-		panic(err)
-	}
 
 	client.execAPI = api.New(client.BaseClient.GetAPI(), client.grpcClient)
 
@@ -201,6 +198,16 @@ func (client *client) InitBlock() {
 	} else {
 		client.SetCurrentBlock(block)
 	}
+	// get main chain calc block hash fork height
+	mainBlockHashForkHeight, err = client.GetForkHeightOnMainChain("ForkBlockHash")
+	if err != nil {
+		panic(err)
+	}
+	mainParaCommitTxForkHeight, err = client.GetForkHeightOnMainChain(pt.ParaX + "-" + pt.ForkCommitTx)
+	if err != nil {
+		panic(err)
+	}
+
 }
 
 // GetStartSeq get startSeq in mainchain
@@ -369,11 +376,11 @@ func (client *client) getLastBlockInfo() (int64, *types.Block, error) {
 
 }
 
-func (client *client) GetBlockHashForkHeightOnMainChain() (int64, error) {
-	ret, err := client.grpcClient.GetFork(context.Background(), &types.ReqKey{Key: []byte("ForkBlockHash")})
+func (client *client) GetForkHeightOnMainChain(key string) (int64, error) {
+	ret, err := client.grpcClient.GetFork(context.Background(), &types.ReqKey{Key: []byte(key)})
 	if err != nil {
 		plog.Error("para get rpc ForkBlockHash fail", "err", err.Error())
-		return -1, err
+		return types.MaxHeight, err
 	}
 
 	return ret.Data, nil
@@ -717,6 +724,8 @@ func (client *client) createBlock(lastBlock *types.Block, txs []*types.Transacti
 	newblock.MainHash = mainBlock.Seq.Hash
 	newblock.MainHeight = mainBlock.Detail.Block.Height
 
+	curMainChainHeight = mainBlock.Detail.Block.Height
+
 	err = client.WriteBlock(lastBlock.StateHash, &newblock, seq)
 
 	plog.Debug("para create new Block", "newblock.ParentHash", common.ToHex(newblock.ParentHash),

plugin/consensus/para/para_test.go

@@ -231,10 +231,10 @@ func TestGetBlockHashForkHeightOnMainChain(t *testing.T) {
 	grpcClient := &typesmocks.Chain33Client{}
 	grpcClient.On("GetFork", mock.Anything, &types.ReqKey{Key: []byte("ForkBlockHash")}).Return(&types.Int64{Data: 1}, errors.New("err")).Once()
 	para.grpcClient = grpcClient
-	_, err := para.GetBlockHashForkHeightOnMainChain()
+	_, err := para.GetForkHeightOnMainChain("ForkBlockHash")
 	assert.NotNil(t, err)
 	grpcClient.On("GetFork", mock.Anything, &types.ReqKey{Key: []byte("ForkBlockHash")}).Return(&types.Int64{Data: 1}, nil).Once()
-	ret, err := para.GetBlockHashForkHeightOnMainChain()
+	ret, err := para.GetForkHeightOnMainChain("ForkBlockHash")
 	assert.Nil(t, err)
 	assert.Equal(t, int64(1), ret)
 

plugin/consensus/para/paracommitmsg.go

@@ -150,13 +150,23 @@ out:
 				"mainHeigt", rsp.mainStatus.Height, "mainlockhash", common.ToHex(rsp.mainStatus.BlockHash),
 				"selfHeight", rsp.selfStatus.Height, "selfHash", common.ToHex(rsp.selfStatus.BlockHash), "sync", isSync)
 
+			if notification == nil {
+				continue
+			}
+
 			//所有节点还没有共识场景或新节点或重启节点catchingUp场景，要等到收到区块高度大于主链共识高度时候发送，在catchingup时候本身共识高度和块高度一起增长
-			if selfConsensusHeight == -1 || (notification != nil && notification[1] > mainConsensHeight) {
+			if selfConsensusHeight == -1 || (notification[1] > mainConsensHeight) {
 				isSync = true
 			}
 
 			//未共识过的小于当前共识高度的区块，可以不参与共识, 如果是新节点，一直等到同步的区块达到了共识高度，才设置同步参与共识
-			if notification != nil && finishHeight < selfConsensusHeight {
+
+			if finishHeight < selfConsensusHeight {
+				finishHeight = selfConsensusHeight
+			}
+
+			// 自共识分叉高度切换场景， 分叉高度前平行链共识高度是-1，分叉高度后，需要重发tx平行链共识高度才能增长
+			if isMainCommitHeightForked() && selfConsensusHeight == -1 && mainConsensHeight > selfConsensusHeight {
 				finishHeight = selfConsensusHeight
 			}
 
@@ -165,7 +175,7 @@ out:
 			//此处也整合了当前consensus height=-1 场景
 			// 需要是<而不是<=, 因为notification[0]被认为是系统起来后已经发送过的
 			nextConsensHeight := selfConsensusHeight + 1
-			if notification != nil && nextConsensHeight < notification[0] {
+			if nextConsensHeight < notification[0] {
 				notification[0] = nextConsensHeight
 				finishHeight = selfConsensusHeight
 				sendingMsgs = nil
@@ -241,7 +251,11 @@ func (client *commitMsgClient) getTxsGroup(txsArr *types.Transactions) (*types.T
 func (client *commitMsgClient) batchCalcTxGroup(notifications []*pt.ParacrossNodeStatus) (*types.Transaction, int, error) {
 	var rawTxs types.Transactions
 	for _, status := range notifications {
-		tx, err := paracross.CreateRawCommitTx4MainChain(status, paracross.GetExecName(), 0)
+		execName := pt.ParaX
+		if isMainCommitHeightForked() {
+			execName = paracross.GetExecName()
+		}
+		tx, err := paracross.CreateRawCommitTx4MainChain(status, execName, 0)
 		if err != nil {
 			plog.Error("para get commit tx", "block height", status.Height)
 			return nil, 0, err
@@ -257,7 +271,11 @@ func (client *commitMsgClient) batchCalcTxGroup(notifications []*pt.ParacrossNod
 }
 
 func (client *commitMsgClient) singleCalcTx(status *pt.ParacrossNodeStatus) (*types.Transaction, error) {
-	tx, err := paracross.CreateRawCommitTx4MainChain(status, paracross.GetExecName(), 0)
+	execName := pt.ParaX
+	if isMainCommitHeightForked() {
+		execName = paracross.GetExecName()
+	}
+	tx, err := paracross.CreateRawCommitTx4MainChain(status, execName, 0)
 	if err != nil {
 		plog.Error("para get commit tx", "block height", status.Height)
 		return nil, err
@@ -331,6 +349,10 @@ func checkTxInMainBlock(targetTx *types.Transaction, detail *types.BlockDetail)
 
 }
 
+func isMainCommitHeightForked() bool {
+	return curMainChainHeight > mainParaCommitTxForkHeight+100
+}
+
 //当前未考虑获取key非常多失败的场景， 如果获取height非常多，block模块会比较大，但是使用完了就释放了
 //如果有必要也可以考虑每次最多取20个一个txgroup，发送共识部分循环获取发送也没问题
 func (client *commitMsgClient) getNodeStatus(start, end int64) ([]*pt.ParacrossNodeStatus, error) {
@@ -529,6 +551,10 @@ out:
 				continue
 			}
 			status.mainStatus = &result
+			//如果没有开启平行链自共识， 采用主链共识, 平行链自共识开启会影响发送tx高度的判断
+			if !isMainCommitHeightForked() {
+				status.selfStatus = status.mainStatus
+			}
 			consensusRst <- &status
 		}
 	}

plugin/consensus/para/paracommitmsg_test.go

@@ -78,6 +78,7 @@ func (s *suiteParaCommitMsg) initEnv(cfg *types.Config, sub *types.ConfigSubModu
 	s.grpcCli = &typesmocks.Chain33Client{}
 
 	s.grpcCli.On("GetFork", mock.Anything, &types.ReqKey{Key: []byte("ForkBlockHash")}).Return(&types.Int64{Data: 1}, nil).Once()
+	s.grpcCli.On("GetFork", mock.Anything, &types.ReqKey{Key: []byte("paracross-ForkParacrossCommitTx")}).Return(&types.Int64{Data: 0}, nil).Once()
 	// GetBlockBySeq return error to stop create's for cycle to request tx
 	s.grpcCli.On("GetBlockBySeq", mock.Anything, mock.Anything).Return(nil, errors.New("quit create"))
 	//data := &types.Int64{1}

plugin/consensus/para/paratxrequest_test.go

@@ -69,6 +69,7 @@ func (s *suiteParaClient) initEnv(cfg *types.Config, sub *types.ConfigSubModule)
 	s.grpcCli = &typesmocks.Chain33Client{}
 
 	s.grpcCli.On("GetFork", mock.Anything, &types.ReqKey{Key: []byte("ForkBlockHash")}).Return(&types.Int64{Data: 1}, nil).Once()
+	s.grpcCli.On("GetFork", mock.Anything, &types.ReqKey{Key: []byte("paracross-ForkParacrossCommitTx")}).Return(&types.Int64{Data: 0}, nil).Once()
 	s.createBlockMock()
 
 	reply := &types.Reply{IsOk: true}

plugin/dapp/paracross/types/type.go

@@ -14,8 +14,10 @@ import (
 
 var (
 	// ParaX paracross exec name
-	ParaX = "paracross"
-	glog  = log.New("module", ParaX)
+	ParaX        = "paracross"
+	glog         = log.New("module", ParaX)
+	// ForkCommitTx main chain support paracross commit tx
+	ForkCommitTx = "ForkParacrossCommitTx"
 )
 
 func init() {
@@ -24,6 +26,7 @@ func init() {
 	types.RegistorExecutor(ParaX, NewType())
 	types.RegisterDappFork(ParaX, "Enable", 0)
 	types.RegisterDappFork(ParaX, "ForkParacrossWithdrawFromParachain", 1298600)
+	types.RegisterDappFork(ParaX, ForkCommitTx, types.MaxHeight)
 }
 
 // GetExecName get para exec name