Merge pull request #46 from persistenceOne/avkr003/0.5.2

Cosmos blocks lag, delegations/redelegations zero amount, ethereum zero address, configuration fixes and improved logging
persistenceOne · Dec 17, 2021 · 6bb3c42 · 6bb3c42
2 parents 08caa3d + edaa61f
commit 6bb3c42
Show file tree

Hide file tree

Showing 7 changed files with 39 additions and 30 deletions.
diff --git a/application/rest/casp/getSignOperation.go b/application/rest/casp/getSignOperation.go
@@ -21,7 +21,7 @@ func GetSignOperation(operationID string) (casp.SignOperationResponse, error) {
 	var response casp.SignOperationResponse
 	client := &http.Client{Transport: &http.Transport{
 		TLSClientConfig: &tls.Config{
-			InsecureSkipVerify: true,
+			InsecureSkipVerify: configuration.GetAppConfig().CASP.TLSInsecureSkipVerify,
 		},
 	}}
 	request, err := http.NewRequest("GET", fmt.Sprintf("%s/casp/api/v1.0/mng/operations/sign/%s", configuration.GetAppConfig().CASP.URL, operationID), nil)

diff --git a/application/rest/casp/postSignData.go b/application/rest/casp/postSignData.go
@@ -41,7 +41,7 @@ func SignData(dataToSign []string, publicKeys []string, description string) (cas
 	responseBody := bytes.NewBuffer(postBody)
 	client := &http.Client{Transport: &http.Transport{
 		TLSClientConfig: &tls.Config{
-			InsecureSkipVerify: true,
+			InsecureSkipVerify: configuration.GetAppConfig().CASP.TLSInsecureSkipVerify,
 		},
 	}}
 	request, err := http.NewRequest("POST", fmt.Sprintf("%s/casp/api/v1.0/mng/vaults/%s/sign", configuration.GetAppConfig().CASP.URL, configuration.GetAppConfig().CASP.VaultID), responseBody)

diff --git a/kafka/handler/ethUnbond.go b/kafka/handler/ethUnbond.go
@@ -106,15 +106,17 @@ ConsumerLoop:
 		}
 
 		for _, unbondMsg := range unbondMsgs {
-			msgBytes, err := m.ProtoCodec.MarshalInterface(unbondMsg)
-			if err != nil {
-				return err
-			}
+			if !unbondMsg.Amount.Amount.LTE(sdk.ZeroInt()) {
+				msgBytes, err := m.ProtoCodec.MarshalInterface(unbondMsg)
+				if err != nil {
+					return err
+				}
 
-			err = utils.ProducerDeliverMessage(msgBytes, utils.MsgUnbond, producer)
-			if err != nil {
-				logging.Error("failed to produce message from: EthUnbond to: MsgUnbond")
-				return err
+				err = utils.ProducerDeliverMessage(msgBytes, utils.MsgUnbond, producer)
+				if err != nil {
+					logging.Error("failed to produce message from: EthUnbond to: MsgUnbond")
+					return err
+				}
 			}
 		}
 		session.MarkMessage(kafkaMsg, "")

diff --git a/kafka/handler/msgDelegate.go b/kafka/handler/msgDelegate.go
@@ -81,17 +81,19 @@ ConsumerLoop:
 			if i == len(validators)-1 {
 				delegateMsg.Amount.Amount = delegateMsg.Amount.Amount.Add(delegationChange)
 			}
-			msgBytes, err := m.ProtoCodec.MarshalInterface(delegateMsg)
-			if err != nil {
-				return err
-			}
+			if !delegateMsg.Amount.Amount.LTE(sdk.ZeroInt()) {
+				msgBytes, err := m.ProtoCodec.MarshalInterface(delegateMsg)
+				if err != nil {
+					return err
+				}
 
-			err = utils.ProducerDeliverMessage(msgBytes, utils.ToTendermint, producer)
-			if err != nil {
-				logging.Error("failed to produce message from: MsgDelegate to ToTendermint")
-				return err
+				err = utils.ProducerDeliverMessage(msgBytes, utils.ToTendermint, producer)
+				if err != nil {
+					logging.Error("failed to produce message from: MsgDelegate to ToTendermint")
+					return err
+				}
+				m.Count++
 			}
-			m.Count++
 		}
 		session.MarkMessage(kafkaMsg, "")
 	}

diff --git a/kafka/handler/redelegate.go b/kafka/handler/redelegate.go
@@ -80,17 +80,20 @@ func (m MsgHandler) HandleRelegate(session sarama.ConsumerGroupSession, claim sa
 		if i == len(validatorSet)-1 {
 			msgRedelegate.Amount.Amount = msgRedelegate.Amount.Amount.Add(redistributeChange)
 		}
-		msgBytes, err := m.ProtoCodec.MarshalInterface(msgRedelegate)
-		if err != nil {
-			return err
-		}
 
-		err = utils.ProducerDeliverMessage(msgBytes, utils.ToTendermint, producer)
-		if err != nil {
-			logging.Error("failed to produce message from topic Redelegate to ToTendermint")
-			return err
+		if !msgRedelegate.Amount.Amount.LTE(sdk.ZeroInt()) {
+			msgBytes, err := m.ProtoCodec.MarshalInterface(msgRedelegate)
+			if err != nil {
+				return err
+			}
+
+			err = utils.ProducerDeliverMessage(msgBytes, utils.ToTendermint, producer)
+			if err != nil {
+				logging.Error("failed to produce message from topic Redelegate to ToTendermint")
+				return err
+			}
+			m.Count++
 		}
-		m.Count++
 	}
 	session.MarkMessage(kafkaMsg, "")
 

diff --git a/tendermint/listener.go b/tendermint/listener.go
@@ -65,7 +65,7 @@ func StartListening(initClientCtx client.Context, chain *relayer.Chain, brokers
 			return
 		}
 
-		if abciInfo.Response.LastBlockHeight > cosmosStatus.LastCheckHeight {
+		if (abciInfo.Response.LastBlockHeight - cosmosStatus.LastCheckHeight) > 5 {
 			processHeight := cosmosStatus.LastCheckHeight + 1
 			logging.Info("Tendermint Block:", processHeight)
 

diff --git a/tendermint/transaction.go b/tendermint/transaction.go
@@ -26,7 +26,8 @@ import (
 )
 
 func handleTxSearchResult(clientCtx client.Context, resultTxs []*tmCoreTypes.ResultTx, kafkaProducer *sarama.SyncProducer, protoCodec *codec.ProtoCodec) error {
-	for _, transaction := range resultTxs {
+	for i, transaction := range resultTxs {
+		logging.Info("Tendermint TX:", transaction.Hash.String(), fmt.Sprintf("(%d)", i+1))
 		err := collectAllWrapAndRevertTxs(clientCtx, transaction)
 		if err != nil {
 			logging.Error("Failed to process tendermint transaction:", transaction.Hash.String())
@@ -111,6 +112,7 @@ func wrapOrRevert(kafkaProducer *sarama.SyncProducer, protoCodec *codec.ProtoCod
 		var ethAddress goEthCommon.Address
 		if validEthMemo {
 			ethAddress = goEthCommon.HexToAddress(tx.Memo)
+			validEthMemo = goEthCommon.Address{}.String() != tx.Memo
 		}
 
 		if tx.Denom == configuration.GetAppConfig().Tendermint.PStakeDenom && validEthMemo && tx.Amount.GTE(sdk.NewInt(configuration.GetAppConfig().Tendermint.MinimumWrapAmount)) {