Bugfix/rc3

block-etl/pkg/chains/IndexMGR.go

@@ -4,10 +4,12 @@ import (
 	"time"
 
 	"github.com/NpoolPlatform/go-service-framework/pkg/logger"
+	"github.com/NpoolPlatform/libent-cruder/pkg/cruder"
 	"github.com/web3eye-io/Web3Eye/block-etl/pkg/chains/eth"
 	"github.com/web3eye-io/Web3Eye/block-etl/pkg/chains/sol"
 	endpointNMCli "github.com/web3eye-io/Web3Eye/nft-meta/pkg/client/v1/endpoint"
 
+	"github.com/web3eye-io/Web3Eye/common/chains"
 	common_eth "github.com/web3eye-io/Web3Eye/common/chains/eth"
 	common_sol "github.com/web3eye-io/Web3Eye/common/chains/sol"
 	"github.com/web3eye-io/Web3Eye/proto/web3eye"
@@ -47,8 +49,8 @@ func init() {
 	}
 
 	// TODO:should be registered
-	pMGR.EndpointChainIDHandlers[basetype.ChainType_Ethereum] = common_eth.GetEndpointChainID
-	pMGR.EndpointChainIDHandlers[basetype.ChainType_Solana] = common_sol.GetEndpointChainID
+	pMGR.EndpointChainIDHandlers[basetype.ChainType_Ethereum] = common_eth.CheckEndpointChainID
+	pMGR.EndpointChainIDHandlers[basetype.ChainType_Solana] = common_sol.CheckEndpointChainID
 }
 
 func GetIndexMGR() *indexMGR {
@@ -57,17 +59,17 @@ func GetIndexMGR() *indexMGR {
 
 func (pmgr *indexMGR) Run(ctx context.Context) {
 	for {
-		pmgr.checkNewEndpoints(ctx, basetype.EndpointState_EndpointDefault)
-		pmgr.checkNewEndpoints(ctx, basetype.EndpointState_EndpointAvaliable)
-		pmgr.checkNewEndpoints(ctx, basetype.EndpointState_EndpointError)
-		pmgr.checkNewEndpoints(ctx, basetype.EndpointState_EndpointUnstable)
-		pmgr.checkAvaliableEndpoints(ctx)
+		pmgr.checkEndpoints(ctx, basetype.EndpointState_EndpointDefault)
+		pmgr.checkEndpoints(ctx, basetype.EndpointState_EndpointAvailable)
+		pmgr.checkEndpoints(ctx, basetype.EndpointState_EndpointError)
+		pmgr.checkEndpoints(ctx, basetype.EndpointState_EndpointUnstable)
+		pmgr.checkAvailableEndpoints(ctx)
 		<-time.NewTicker(UpdateInterval).C
 	}
 }
 
 // check for the newly created endpoints
-func (pmgr *indexMGR) checkNewEndpoints(ctx context.Context, state basetype.EndpointState) {
+func (pmgr *indexMGR) checkEndpoints(ctx context.Context, state basetype.EndpointState) {
 	conds := &endpoint.Conds{
 		State: &web3eye.Uint32Val{
 			Op:    "eq",
@@ -98,12 +100,13 @@ func (pmgr *indexMGR) checkNewEndpoints(ctx context.Context, state basetype.Endp
 
 			chainID, err := handler(ctx, info.Address)
 			if err != nil {
+				logger.Sugar().Errorf("endpoint %v not available, err: %v", info.ID, err)
 				info.State = basetype.EndpointState_EndpointError
 				return
 			}
 
 			info.ChainID = chainID
-			info.State = basetype.EndpointState_EndpointAvaliable
+			info.State = basetype.EndpointState_EndpointAvailable
 		}()
 
 		updateInfos = append(updateInfos, &endpoint.EndpointReq{
@@ -134,11 +137,11 @@ func (pmgr *indexMGR) checkNewEndpoints(ctx context.Context, state basetype.Endp
 }
 
 // check erver chantype-chainid available endpoints and update it to indexer
-func (pmgr *indexMGR) checkAvaliableEndpoints(ctx context.Context) {
+func (pmgr *indexMGR) checkAvailableEndpoints(ctx context.Context) {
 	conds := &endpoint.Conds{
 		State: &web3eye.Uint32Val{
-			Op:    "eq",
-			Value: uint32(basetype.EndpointState_EndpointAvaliable),
+			Op:    cruder.EQ,
+			Value: uint32(basetype.EndpointState_EndpointAvailable),
 		},
 	}
 
@@ -159,6 +162,18 @@ func (pmgr *indexMGR) checkAvaliableEndpoints(ctx context.Context) {
 			endpointGroups[info.ChainType][info.ChainID] = []string{}
 		}
 		endpointGroups[info.ChainType][info.ChainID] = append(endpointGroups[info.ChainType][info.ChainID], info.Address)
+		if info.RPS == 0 {
+			info.RPS = 1
+		}
+
+		err = chains.GetEndpintIntervalMGR().GoAheadEndpoint(&chains.EndpointInterval{
+			Address:     info.Address,
+			MinInterval: time.Second / time.Duration(info.RPS),
+			MaxInterval: time.Minute,
+		})
+		if err != nil {
+			logger.Sugar().Warnw("checkAvailableEndpoints", "Msg", "failed to put endpoints to redis", "Error", err)
+		}
 	}
 
 	// check if have no endpoints,will be stop

block-etl/pkg/chains/eth/price.go

@@ -35,7 +35,7 @@ func (e *EthIndexer) IndexOrder(ctx context.Context, logs []*types.Log) ([]*Cont
 		}
 	}
 
-	_, err := orderNMCli.CreateOrders(ctx, &order.CreateOrdersRequest{Infos: ordersReq})
+	_, err := orderNMCli.UpsertOrders(ctx, &order.UpsertOrdersRequest{Infos: ordersReq})
 	if err != nil {
 		return nil, fmt.Errorf("failed store orders to db,err: %v", err)
 	}

block-etl/pkg/chains/eth/task.go

@@ -2,6 +2,7 @@ package eth
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/NpoolPlatform/go-service-framework/pkg/logger"
@@ -12,9 +13,9 @@ import (
 )
 
 const (
-	CheckTopicInterval     = time.Second * 10
+	CheckTopicInterval     = time.Second * 5
 	FindContractCreator    = false
-	redisExpireDefaultTime = time.Second * 10
+	redisExpireDefaultTime = time.Second * 5
 )
 
 type EthIndexer struct {
@@ -43,60 +44,68 @@ func (e *EthIndexer) IndexBlock(ctx context.Context, taskBlockNum chan uint64) {
 	for {
 		select {
 		case num := <-taskBlockNum:
+			logger.Sugar().Infow("IndexBlock", "BlockNum", num, "DebugMsg", "start to parse")
 			block, err := e.CheckBlock(ctx, num)
 			if err != nil {
 				logger.Sugar().Errorw("IndexBlock", "BlockNum", num, "Error", err)
 				continue
 			}
 
-			if block.ParseState == basetype.BlockParseState_BlockTypeFinish {
+			if block.ParseState != basetype.BlockParseState_BlockTypeStart {
 				continue
 			}
 
 			err = func() error {
+				err := e.checkOkEndpoints()
+				if err != nil {
+					return err
+				}
+				logger.Sugar().Infow("IndexBlock", "BlockNum", num, "DebugMsg", "start IndexBlockLogs")
+
 				blockLogs, err := e.IndexBlockLogs(ctx, block.BlockNumber)
 				if err != nil {
 					return err
 				}
+				logger.Sugar().Infow("IndexBlock", "BlockNum", num, "DebugMsg", "start IndexTransfer")
 
 				filteredT1, err := e.IndexTransfer(ctx, blockLogs.TransferLogs, block.BlockTime)
 				if err != nil {
 					return err
 				}
+				logger.Sugar().Infow("IndexBlock", "BlockNum", num, "DebugMsg", "start IndexToken")
 
 				contractT1, err := e.IndexToken(ctx, filteredT1)
 				if err != nil {
 					return err
 				}
+				logger.Sugar().Infow("IndexBlock", "BlockNum", num, "DebugMsg", "start IndexContract")
 
 				err = e.IndexContract(ctx, contractT1, FindContractCreator)
 				if err != nil {
 					return err
 				}
+				logger.Sugar().Infow("IndexBlock", "BlockNum", num, "DebugMsg", "start IndexOrder")
 
 				contractT2, err := e.IndexOrder(ctx, blockLogs.OrderLogs)
 				if err != nil {
 					return err
 				}
-
+				logger.Sugar().Infow("IndexBlock", "BlockNum", num, "DebugMsg", "start order IndexContract")
 				err = e.IndexContract(ctx, contractT2, FindContractCreator)
 				if err != nil {
 					return err
 				}
 				return nil
 			}()
 
-			if err != nil {
-				logger.Sugar().Errorw("IndexBlock", "BlockNum", num, "Error", err)
-			}
-
 			remark := ""
 			parseState := basetype.BlockParseState_BlockTypeFinish
 			if err != nil {
+				logger.Sugar().Errorw("IndexBlock", "BlockNum", num, "Error", err)
 				remark = err.Error()
 				parseState = basetype.BlockParseState_BlockTypeFailed
 			}
-
+			logger.Sugar().Infow("IndexBlock", "BlockNum", num, "DebugMsg", "start order UpdateBlock")
 			_, err = blockNMCli.UpdateBlock(ctx, &blockProto.UpdateBlockRequest{
 				Info: &blockProto.BlockReq{
 					ID:         &block.ID,
@@ -117,11 +126,17 @@ func (e *EthIndexer) OnNoAvalibleEndpoints(event func()) {
 	e.ONAEEvents = append(e.ONAEEvents, event)
 }
 
-func (e *EthIndexer) UpdateEndpoints(endpoints []string) {
-	e.OkEndpoints = endpoints
+func (e *EthIndexer) checkOkEndpoints() error {
 	if len(e.OkEndpoints) == 0 {
 		for _, v := range e.ONAEEvents {
 			v()
 		}
+		return fmt.Errorf("have no available endpoints")
 	}
+	return nil
+}
+
+func (e *EthIndexer) UpdateEndpoints(endpoints []string) {
+	e.OkEndpoints = endpoints
+	_ = e.checkOkEndpoints()
 }