Merge pull request #204 from meshplus/fix/fix-direct-v1.6

fix(direct):fix start in direct mode

Author: jzhe886

api/server.go

@@ -159,23 +159,23 @@ func (g *Server) getAppchain(c *gin.Context) {
 		return
 	}
 
+	if !ackMsg.Payload.Ok {
+		res.Data = ackMsg.Payload.Data
+		c.JSON(http.StatusInternalServerError, res)
+		return
+	}
+
 	g.handleAckAppchain(c, ackMsg)
 }
 
 func (g *Server) handleAckAppchain(c *gin.Context, msg *peerproto.Message) {
-	app := appchainmgr.Appchain{}
-	if err := json.Unmarshal(msg.Payload.Data, &app); err != nil {
-		g.logger.Error(err)
-		return
-	}
-
 	res := &response{}
 
 	switch msg.Type {
 	case peerproto.Message_APPCHAIN_REGISTER:
-		res.Data = []byte(fmt.Sprintf("appchain register successfully, id is %s\n", app.ID))
+		res.Data = []byte(fmt.Sprintf("appchain register successfully, appchain is %s\n", string(msg.Payload.Data)))
 	case peerproto.Message_APPCHAIN_UPDATE:
-		res.Data = []byte(fmt.Sprintf("appchain update successfully, id is %s\n", app.ID))
+		res.Data = []byte(fmt.Sprintf("appchain update successfully, appchain is %s\n", string(msg.Payload.Data)))
 	case peerproto.Message_APPCHAIN_GET:
 		res.Data = msg.Payload.Data
 	}

cmd/pier/client/appchain_pier.go

@@ -219,7 +219,7 @@ func savePierAppchain(ctx *cli.Context, path string) error {
 	validatorsPath := ctx.String("validators")
 	consensusType := ctx.String("consensusType")
 
-	url, err := getURL(ctx, fmt.Sprintf("%s?pier_id=%s", path, pier))
+	url, err := getURL(ctx, fmt.Sprintf("%s?pier_id=%s", GetAppchainUrl, pier))
 	if err != nil {
 		return err
 	}

go.mod

@@ -4,10 +4,10 @@ go 1.13
 
 require (
 	github.com/Rican7/retry v0.1.0
-	github.com/btcsuite/btcd v0.21.0-beta
+	github.com/btcsuite/btcd v0.20.1-beta
 	github.com/cbergoon/merkletree v0.2.0
 	github.com/fatih/color v1.9.0
-	github.com/fsnotify/fsnotify v1.4.9
+	github.com/fsnotify/fsnotify v1.4.7
 	github.com/gin-gonic/gin v1.6.3
 	github.com/gobuffalo/packd v1.0.0
 	github.com/gobuffalo/packr v1.30.1

internal/app/pier.go

@@ -123,6 +123,11 @@ func NewPier(repoRoot string, config *repo.Config) (*Pier, error) {
 		if err != nil {
 			return nil, fmt.Errorf("cryptor create: %w", err)
 		}
+		pierHAConstructor, err := agency.GetPierHAConstructor(config.HA.Mode)
+		if err != nil {
+			return nil, fmt.Errorf("pier ha constructor not found")
+		}
+		pierHA = pierHAConstructor(nil, addr.String())
 
 		meta = &pb.Interchain{}
 		lite = &direct_lite.MockLite{}

internal/appchain/handle.go

@@ -40,18 +40,8 @@ func (mgr *Manager) handleMessage(s network.Stream, msg *peerproto.Message) {
 		mgr.logger.Error(err)
 	}
 
-	appchainRes := &appchainmgr.Appchain{}
-	if err := json.Unmarshal(res, appchainRes); err != nil {
-		mgr.logger.Error(err)
-		return
-	}
-
 	mgr.logger.WithFields(logrus.Fields{
-		"type":           msg.Type,
-		"from_id":        appchainRes.ID,
-		"name":           appchainRes.Name,
-		"desc":           appchainRes.Desc,
-		"chain_type":     appchainRes.ChainType,
-		"consensus_type": appchainRes.ConsensusType,
+		"type":    msg.Type,
+		"from_id": string(res),
 	}).Info("Handle appchain message")
 }

internal/exchanger/direct_handler.go

@@ -44,15 +44,8 @@ func (pool *Pool) get(index uint64) *model.WrappedIBTP {
 }
 
 func (ex *Exchanger) feedIBTP(wIbtp *model.WrappedIBTP) {
-	var pool *Pool
-	ibtp := wIbtp.Ibtp
-	act, loaded := ex.ibtps.Load(ibtp.From)
-	if !loaded {
-		pool = NewPool()
-		ex.ibtps.Store(ibtp.From, pool)
-	} else {
-		pool = act.(*Pool)
-	}
+	act, loaded := ex.ibtps.LoadOrStore(wIbtp.Ibtp.From, NewPool())
+	pool := act.(*Pool)
 	pool.feed(wIbtp)
 
 	if !loaded {
@@ -63,53 +56,47 @@ func (ex *Exchanger) feedIBTP(wIbtp *model.WrappedIBTP) {
 				}
 			}()
 			inMeta := ex.exec.QueryInterchainMeta()
+			idx := inMeta[wIbtp.Ibtp.From]
 			for wIbtp := range pool.ch {
-				ibtp := wIbtp.Ibtp
-				idx := inMeta[ibtp.From]
-				if ibtp.Index <= idx {
-					pool.delete(ibtp.Index)
-					ex.logger.Warnf("ignore ibtp with invalid index: %d", ibtp.Index)
+				if wIbtp.Ibtp.Index <= idx {
+					pool.delete(wIbtp.Ibtp.Index)
+					ex.logger.Warnf("ignore ibtp with invalid index: %d", wIbtp.Ibtp.Index)
 					continue
 				}
-				if idx+1 == ibtp.Index {
-					ex.processIBTP(wIbtp)
-					pool.delete(ibtp.Index)
-					index := ibtp.Index + 1
+				if idx+1 == wIbtp.Ibtp.Index {
+					ex.processIBTP(wIbtp, pool)
+					index := wIbtp.Ibtp.Index + 1
 					wIbtp := pool.get(index)
 					for wIbtp != nil {
-						ex.processIBTP(wIbtp)
-						pool.delete(wIbtp.Ibtp.Index)
+						ex.processIBTP(wIbtp, pool)
 						index++
 						wIbtp = pool.get(index)
 					}
+					idx = index - 1
 				} else {
 					pool.put(wIbtp)
 				}
+
 			}
 		}(pool)
 	}
 }
 
-func (ex *Exchanger) processIBTP(wIbtp *model.WrappedIBTP) {
+func (ex *Exchanger) processIBTP(wIbtp *model.WrappedIBTP, pool *Pool) {
 	receipt, err := ex.exec.ExecuteIBTP(wIbtp)
 	if err != nil {
 		ex.logger.Errorf("Execute ibtp error: %s", err.Error())
 		return
 	}
 	ex.postHandleIBTP(wIbtp.Ibtp.From, receipt)
 	ex.sendIBTPCounter.Inc()
+	pool.delete(wIbtp.Ibtp.Index)
 }
 
-func (ex *Exchanger) feedReceipt(receipt *pb.IBTP) {
-	var pool *Pool
-	act, loaded := ex.ibtps.Load(receipt.To)
-	if !loaded {
-		pool = NewPool()
-		ex.ibtps.Store(receipt.To, pool)
-	} else {
-		pool = act.(*Pool)
-	}
-	pool.feed(&model.WrappedIBTP{Ibtp: receipt, IsValid: true})
+func (ex *Exchanger) feedReceipt(ibtp *pb.IBTP) {
+	act, loaded := ex.ibtps.LoadOrStore(ibtp.From, NewPool())
+	pool := act.(*Pool)
+	pool.feed(&model.WrappedIBTP{Ibtp: ibtp, IsValid: true})
 
 	if !loaded {
 		go func(pool *Pool) {
@@ -119,26 +106,25 @@ func (ex *Exchanger) feedReceipt(receipt *pb.IBTP) {
 				}
 			}()
 			callbackMeta := ex.exec.QueryCallbackMeta()
+			idx := callbackMeta[ibtp.To]
 			for wIbtp := range pool.ch {
-				ibtp := wIbtp.Ibtp
-				if ibtp.Index <= callbackMeta[ibtp.To] {
-					pool.delete(ibtp.Index)
+				if ibtp.Index <= idx {
+					pool.delete(wIbtp.Ibtp.Index)
 					ex.logger.Warn("ignore ibtp with invalid index")
 					continue
 				}
-				if callbackMeta[ibtp.To]+1 == ibtp.Index {
-					ex.processIBTP(wIbtp)
-					pool.delete(ibtp.Index)
+				if idx+1 == ibtp.Index {
+					ex.processIBTP(wIbtp, pool)
 					index := ibtp.Index + 1
 					wIbtp := pool.get(index)
 					for wIbtp != nil {
-						ibtp := wIbtp.Ibtp
 						receipt, _ := ex.exec.ExecuteIBTP(wIbtp)
-						ex.postHandleIBTP(ibtp.From, receipt)
-						pool.delete(ibtp.Index)
+						ex.postHandleIBTP(wIbtp.Ibtp.From, receipt)
+						pool.delete(wIbtp.Ibtp.Index)
 						index++
 						wIbtp = pool.get(index)
 					}
+					idx = index - 1
 				} else {
 					pool.put(wIbtp)
 				}

internal/exchanger/exchanger.go

@@ -329,7 +329,7 @@ func (ex *Exchanger) sendIBTP(ibtp *pb.IBTP) error {
 			ex.logger.Panic(err)
 		}
 	}
-	entry.Info("Send ibtp success from monitor")
+	//entry.Info("Send ibtp success from monitor")
 	return nil
 }