fix: add missing mutex lock.

relayer/chains/icon/icon_chain_processor.go

@@ -76,9 +76,9 @@
 	prevNetworkSectionHash []byte
 }
 
 func NewIconChainProcessor(log *zap.Logger, provider *IconProvider, metrics *processor.PrometheusMetrics, heightSnapshot chan struct{}) *IconChainProcessor {
 	return &IconChainProcessor{
 		log:                  log.With(zap.String("chain_name", provider.ChainName()), zap.String("chain_id", provider.ChainId())),
 		chainProvider:        provider,
 		latestClientState:    make(latestClientState),
 		connectionStateCache: make(processor.ConnectionStateCache),
@@ -86,7 +86,7 @@
 		connectionClients:    make(map[string]string),
 		channelConnections:   make(map[string]string),
 		metrics:              metrics,
 		heightSnapshotChan:   heightSnapshot,
 	}
 }
 
@@ -152,7 +152,7 @@
 	}
 
 	// start_query_cycle
 	icp.log.Debug("Starting query cycle")
 	err := icp.monitoring(ctx, &persistence)
 	return err
 }
@@ -171,12 +171,12 @@
 	return snapshotHeight
 }
 
 func (icp *IconChainProcessor) getLastSavedHeight() int {
 	snapshotHeight, err := rlycommon.LoadSnapshotHeight(icp.Provider().ChainId())
 	if err != nil || snapshotHeight < 0 {
 		return 0
 	}
 	return snapshotHeight
 }
 
 func (icp *IconChainProcessor) initializeConnectionState(ctx context.Context) error {
@@ -198,9 +198,9 @@
 			CounterpartyClientID: c.Counterparty.ClientId,
 		}] = c.State == conntypes.OPEN
 
 		icp.log.Debug("Found open connection",
 			zap.String("client-id ", c.ClientId),
 			zap.String("connection-id ", c.Id),
 		)
 	}
 	return nil
@@ -232,11 +232,11 @@
 			CounterpartyPortID:    ch.Counterparty.PortId,
 		}] = ch.State == chantypes.OPEN
 
 		icp.log.Debug("Found open channel",
 			zap.String("channel-id", ch.ChannelId),
 			zap.String("port-id ", ch.PortId),
 			zap.String("counterparty-channel-id", ch.Counterparty.ChannelId),
 			zap.String("counterparty-port-id", ch.Counterparty.PortId))
 	}
 
 	return nil
@@ -285,9 +285,8 @@
 			return err
 		}
 	}
-	// }
 
 	icp.log.Info("Start to query from height", zap.Int64("height", processedheight))
 	// subscribe to monitor block
 	ctxMonitorBlock, cancelMonitorBlock := context.WithCancel(ctx)
 	reconnect()
@@ -297,7 +296,7 @@
 	icp.firstTime = true
 
 	blockReq := &types.BlockRequest{
-		Height:       types.NewHexInt(int64(icp.chainProvider.PCfg.StartHeight)),
+		Height:       types.NewHexInt(int64(processedheight)),
 		EventFilters: GetMonitorEventFilters(icp.chainProvider.PCfg.IbcHandlerAddress),
 	}
 
@@ -309,8 +308,8 @@
 		case err := <-errCh:
 			return err
 
 		case <-icp.heightSnapshotChan:
 			icp.SnapshotHeight(icp.getHeightToSave(int64(icp.latestBlock.Height)))
 
 		case <-reconnectCh:
 			cancelMonitorBlock()
@@ -318,7 +317,7 @@
 
 			go func(ctx context.Context, cancel context.CancelFunc) {
 				blockReq.Height = types.NewHexInt(processedheight)
 				icp.log.Debug("Try to reconnect from", zap.Int64("height", processedheight))
 				err := icp.chainProvider.client.MonitorBlock(ctx, blockReq, func(conn *websocket.Conn, v *types.BlockNotification) error {
 					if !errors.Is(ctx.Err(), context.Canceled) {
 						btpBlockNotifCh <- v
@@ -327,10 +326,10 @@
 				}, func(conn *websocket.Conn) {
 				}, func(conn *websocket.Conn, err error) {})
 				if err != nil {
 					ht := icp.getHeightToSave(processedheight)
 					if ht != icp.getLastSavedHeight() {
 						icp.SnapshotHeight(ht)
 					}
 					if errors.Is(err, context.Canceled) {
 						return
 					}
@@ -346,16 +345,16 @@
 				err := icp.verifyBlock(ctx, br.Header)
 				if err != nil {
 					reconnect()
 					icp.log.Warn("Failed to verify BTP Block",
 						zap.Int64("height", br.Height),
 						zap.Error(err),
 					)
 					break
 				}
 
 				icp.log.Debug("Verified block ",
 					zap.Int64("height", int64(processedheight)))
 
 				icp.latestBlock = provider.LatestBlock{
 					Height: uint64(processedheight),
 				}
@@ -368,7 +367,7 @@
 				}
 
 				ibcHeaderCache[uint64(br.Height)] = br.Header
 				icp.log.Debug("Queried block ",
 					zap.Int64("height", br.Height))
 				err = icp.handlePathProcessorUpdate(ctx, br.Header, ibcMessageCache, ibcHeaderCache.Clone())
 				if err != nil {
@@ -379,6 +378,9 @@
 					break
 				}
 				time.Sleep(10 * time.Millisecond)
+				if icp.firstTime {
+					time.Sleep(4000 * time.Millisecond)
+				}
 				icp.firstTime = false
 				if br = nil; len(btpBlockRespCh) > 0 {
 					br = <-btpBlockRespCh
@@ -400,7 +402,7 @@
 					if err != nil {
 						return err
 					} else if height != processedheight+i {
-						icp.log.Warn("Reconnect: missing block notification ",
+						icp.log.Warn("Reconnect: missing block notification",
 							zap.Int64("got", height),
 							zap.Int64("expected", processedheight+i),
 						)
@@ -473,20 +475,20 @@
 	}
 }
 
 func (icp *IconChainProcessor) getHeightToSave(height int64) int {
 	retryAfter := icp.Provider().ProviderConfig().GetFirstRetryBlockAfter()
 	ht := int(height - int64(retryAfter))
 	if ht < 0 {
 		return 0
 	}
 	return ht
 }
 
 func (icp *IconChainProcessor) SnapshotHeight(height int) {
 	icp.log.Info("Save height for snapshot", zap.Int("height", height))
 	err := rlycommon.SnapshotHeight(icp.Provider().ChainId(), height)
 	if err != nil {
 		icp.log.Warn("Failed saving height snapshot for height", zap.Int("height", height))
 	}
 }
 
@@ -545,8 +547,8 @@
 	icp.verifier.nextProofContext = header.Validators
 	icp.verifier.verifiedHeight = int64(header.Height())
 	icp.verifier.prevNetworkSectionHash = types.NewNetworkSection(header.Header).Hash()
 	icp.log.Debug("Verified block ",
 		zap.Uint64("height", header.Height()))
 	return nil
 }
 
@@ -623,8 +625,8 @@
 						request.err = errors.Wrapf(err, "event.UnmarshalFromBytes: %v", err)
 						return
 					}
 					icp.log.Info("Detected eventlog ", zap.Int64("height", request.height),
 						zap.String("eventlog", IconCosmosEventMap[string(el.Indexed[0])]))
 					eventlogs = append(eventlogs, el)
 				}
 
@@ -650,7 +652,7 @@
 			request.response.IsProcessed = processed
 			return
 		}
 		request.err = errors.Wrapf(err, "Failed to get btp header: %v", err)
 		return
 	}
 	request.response.Header = NewIconIBCHeader(btpHeader, validators, int64(btpHeader.MainHeight))
@@ -697,6 +699,7 @@
 	if state, ok := icp.latestClientState[clientID]; ok {
 		return state, nil
 	}
+
 	cs, err := icp.chainProvider.QueryClientStateWithoutProof(ctx, int64(icp.latestBlock.Height), clientID)
 	if err != nil {
 		return provider.ClientState{}, err

relayer/chains/icon/provider.go

@@ -191,20 +191,20 @@
 //ChainProvider Methods
 
 func (icp *IconProvider) Init(ctx context.Context) error {
 	// if _, err := os.Stat(icp.PCfg.Keystore); err != nil {
 	// 	return err
 	// }
 
 	// ksByte, err := os.ReadFile(icp.PCfg.Keystore)
 	// if err != nil {
 	// 	return err
 	// }
 
 	// wallet, err := wallet.NewFromKeyStore(ksByte, []byte(icp.PCfg.Password))
 	// if err != nil {
 	// 	return err
 	// }
 	// icp.AddWallet(wallet)
 	return nil
 }
 
@@ -225,7 +225,7 @@
 		return nil, fmt.Errorf("Blockinterval cannot be empty in Icon config")
 	}
 
-	trustingBlockPeriod := uint64(dstTrustingPeriod) / (icp.PCfg.BlockInterval * uint64(common.NanosecondRatio))
+	trustingBlockPeriod := uint64(dstTrustingPeriod) / (icp.PCfg.BlockInterval * uint64(common.NanoToMilliRatio))
 
 	return &icon.ClientState{
 		// In case of Icon: Trusting Period is block Difference // see: light.proto in ibc-integration
@@ -262,7 +262,7 @@
 
 func (icp *IconProvider) ConnectionProof(ctx context.Context, msgOpenAck provider.ConnectionInfo, height uint64) (provider.ConnectionProof, error) {
 
 	connState, err := icp.QueryConnection(ctx, int64(msgOpenAck.Height), msgOpenAck.ConnID)
 	if err != nil {
 		return provider.ConnectionProof{}, err
 	}
@@ -273,16 +273,16 @@
 }
 
 func (icp *IconProvider) ChannelProof(ctx context.Context, msg provider.ChannelInfo, height uint64) (provider.ChannelProof, error) {
 
 	channelResult, err := icp.QueryChannel(ctx, int64(msg.Height), msg.ChannelID, msg.PortID)
 	if err != nil {
 		return provider.ChannelProof{}, nil
 	}
 	return provider.ChannelProof{
 		Proof:       channelResult.Proof,
 		ProofHeight: channelResult.ProofHeight,
 		Ordering:    chantypes.Order(channelResult.Channel.GetOrdering()),
 		Version:     channelResult.Channel.Version,
 	}, nil
 }
 
@@ -332,7 +332,7 @@
 	}
 	return provider.PacketProof{
 		Proof:       packetAckResponse.Proof,
 		ProofHeight: packetAckResponse.ProofHeight,
 	}, nil
 
 }
@@ -435,7 +435,7 @@
 }
 
 func (icp *IconProvider) Key() string {
 	return icp.PCfg.Keystore
 }
 
 func (icp *IconProvider) Wallet() (module.Wallet, error) {
@@ -443,7 +443,7 @@
 }
 
 func (icp *IconProvider) Address() (string, error) {
 	return icp.ShowAddress(icp.PCfg.Keystore)
 }
 
 func (icp *IconProvider) Timeout() string {

relayer/chains/wasm/query.go

@@ -11,6 +11,7 @@
 	"time"
 
 	wasmtypes "github.com/CosmWasm/wasmd/x/wasm/types"
+	"github.com/avast/retry-go/v4"
 	abci "github.com/cometbft/cometbft/abci/types"
 	rpcclient "github.com/cometbft/cometbft/rpc/client"
 	tmtypes "github.com/cometbft/cometbft/types"
@@ -19,6 +20,7 @@
 	"github.com/cosmos/gogoproto/proto"
 	tmclient "github.com/cosmos/ibc-go/v7/modules/light-clients/07-tendermint"
 	"github.com/icon-project/IBC-Integration/libraries/go/common/icon"
+	"go.uber.org/zap"
 
 	querytypes "github.com/cosmos/cosmos-sdk/types/query"
 	bankTypes "github.com/cosmos/cosmos-sdk/x/bank/types"
@@ -313,33 +315,45 @@
 }
 
 func (ap *WasmProvider) QueryClientConsensusState(ctx context.Context, chainHeight int64, clientid string, clientHeight ibcexported.Height) (*clienttypes.QueryConsensusStateResponse, error) {
 
 	consensusStateParam, err := types.NewConsensusStateByHeight(clientid, uint64(clientHeight.GetRevisionHeight())).Bytes()
 	consensusState, err := ap.QueryIBCHandlerContractProcessed(ctx, consensusStateParam)
 	if err != nil {
 		return nil, err
 	}
 
 	cdc := codec.NewProtoCodec(ap.Cdc.InterfaceRegistry)
 	csState, err := clienttypes.UnmarshalConsensusState(cdc, consensusState)
 	if err != nil {
 		return nil, err
 	}
 
 	anyConsensusState, err := clienttypes.PackConsensusState(csState)
 	if err != nil {
 		return nil, err
 	}
 	return clienttypes.NewQueryConsensusStateResponse(anyConsensusState, nil, clienttypes.NewHeight(0, uint64(chainHeight))), nil
 }
 
-func (ap *WasmProvider) QueryIBCHandlerContract(ctx context.Context, param wasmtypes.RawContractMessage) (*wasmtypes.QuerySmartContractStateResponse, error) {
-	done := ap.SetSDKContext()
-	defer done()
-	return ap.QueryClient.SmartContractState(ctx, &wasmtypes.QuerySmartContractStateRequest{
-		Address:   ap.PCfg.IbcHandlerAddress,
-		QueryData: param,
-	})
+func (ap *WasmProvider) QueryIBCHandlerContract(ctx context.Context, param wasmtypes.RawContractMessage) (op *wasmtypes.QuerySmartContractStateResponse, err error) {
+	return op, retry.Do(func() error {
+		done := ap.SetSDKContext()
+		defer done()
+		op, err = ap.QueryClient.SmartContractState(ctx, &wasmtypes.QuerySmartContractStateRequest{
+			Address:   ap.PCfg.IbcHandlerAddress,
+			QueryData: param,
+		})
+		return err
+	}, retry.Context(ctx), retry.Attempts(latestHeightQueryRetries), retry.Delay(50*time.Millisecond), retry.LastErrorOnly(true), retry.OnRetry(func(n uint, err error) {
+		ap.log.Error(
+			"Failed to query",
+			zap.Uint("attempt", n+1),
+			zap.Uint("max_attempts", latestHeightQueryRetries),
+			zap.Any("Param", param),
+			zap.Error(err),
+		)
+	}))
+
 }
 
 func (ap *WasmProvider) QueryIBCHandlerContractProcessed(ctx context.Context, param wasmtypes.RawContractMessage) ([]byte, error) {
@@ -492,6 +506,8 @@
 }
 
 func (ap *WasmProvider) QueryWasmProof(ctx context.Context, storageKey []byte, height int64) ([]byte, error) {
+	done := ap.SetSDKContext()
+	defer done()
 	ibcAddr, err := sdk.AccAddressFromBech32(ap.PCfg.IbcHandlerAddress)
 	if err != nil {
 		return nil, err

relayer/chains/wasm/wasm_chain_processor.go

@@ -66,7 +66,7 @@
 	Header *types.LightBlock
 }
 
 func NewWasmChainProcessor(log *zap.Logger, provider *WasmProvider, metrics *processor.PrometheusMetrics, heightSnapshot chan struct{}) *WasmChainProcessor {
 	return &WasmChainProcessor{
 		log:                  log.With(zap.String("chain_name", provider.ChainName()), zap.String("chain_id", provider.ChainId())),
 		chainProvider:        provider,
@@ -76,7 +76,7 @@
 		connectionClients:    make(map[string]string),
 		channelConnections:   make(map[string]string),
 		metrics:              metrics,
 		heightSnapshotChan:   heightSnapshot,
 	}
 }
 
@@ -90,6 +90,7 @@
 	defaultMinQueryLoopDuration      = 1 * time.Second
 	defaultBalanceUpdateWaitDuration = 60 * time.Second
 	inSyncNumBlocksThreshold         = 2
+	MaxBlockFetch                    = 100
 )
 
 // latestClientState is a map of clientID to the latest clientInfo for that client.
@@ -176,7 +177,7 @@
 // clientState will return the most recent client state if client messages
 // have already been observed for the clientID, otherwise it will query for it.
 func (ccp *WasmChainProcessor) clientState(ctx context.Context, clientID string) (provider.ClientState, error) {
-	if state, ok := ccp.latestClientState[clientID]; ok && state.TrustingPeriod > 0 {
+	if state, ok := ccp.latestClientState[clientID]; ok {
 		return state, nil
 	}
 	cs, err := ccp.chainProvider.QueryClientState(ctx, int64(ccp.latestBlock.Height), clientID)
@@ -221,7 +222,7 @@
 func (ccp *WasmChainProcessor) Run(ctx context.Context, initialBlockHistory uint64) error {
 	// this will be used for persistence across query cycle loop executions
 	persistence := queryCyclePersistence{
-		minQueryLoopDuration:      time.Duration(ccp.chainProvider.PCfg.BlockInterval * uint64(common.NanosecondRatio)),
+		minQueryLoopDuration:      time.Duration(ccp.chainProvider.PCfg.BlockInterval * uint64(common.NanoToMilliRatio)),
 		lastBalanceUpdate:         time.Unix(0, 0),
 		balanceUpdateWaitDuration: defaultBalanceUpdateWaitDuration,
 	}
@@ -293,8 +294,8 @@
 		select {
 		case <-ctx.Done():
 			return nil
 		case <-ccp.heightSnapshotChan:
 			ccp.SnapshotHeight(ccp.getHeightToSave(persistence.latestHeight))
 		case <-ticker.C:
 			ticker.Reset(persistence.minQueryLoopDuration)
 		}
@@ -317,11 +318,11 @@
 			CounterpartyConnID:   c.Counterparty.ConnectionId,
 			CounterpartyClientID: c.Counterparty.ClientId,
 		}] = c.State == conntypes.OPEN
 
 		ccp.log.Debug("Found open connection",
 			zap.String("client-id ", c.ClientId),
 			zap.String("connection-id ", c.Id),
 		)
 	}
 	return nil
 }
@@ -350,11 +351,11 @@
 			CounterpartyChannelID: ch.Counterparty.ChannelId,
 			CounterpartyPortID:    ch.Counterparty.PortId,
 		}] = ch.State == chantypes.OPEN
 		ccp.log.Debug("Found open channel",
 			zap.String("channel-id", ch.ChannelId),
 			zap.String("port-id ", ch.PortId),
 			zap.String("counterparty-channel-id", ch.Counterparty.ChannelId),
 			zap.String("counterparty-port-id", ch.Counterparty.PortId))
 	}
 	return nil
 }
@@ -368,16 +369,16 @@
 			zap.Uint("attempts", latestHeightQueryRetries),
 			zap.Error(err),
 		)
 
 		// TODO: Save height when node status is false?
 		// ccp.SnapshotHeight(ccp.getHeightToSave(status.SyncInfo.LatestBlockHeight))
 		return nil
 	}
 
 	persistence.latestHeight = status.SyncInfo.LatestBlockHeight
 	// ccp.chainProvider.setCometVersion(ccp.log, status.NodeInfo.Version)
 
 	ccp.log.Debug("Queried latest height",
 		zap.Int64("latest_height", persistence.latestHeight),
 	)
 
@@ -410,7 +411,14 @@
 	chainID := ccp.chainProvider.ChainId()
 	var latestHeader provider.IBCHeader
 
-	for i := persistence.latestQueriedBlock + 1; i <= persistence.latestHeight; i++ {
+	syncUpHeight := func() int64 {
+		if persistence.latestHeight-persistence.latestQueriedBlock > MaxBlockFetch {
+			return persistence.latestQueriedBlock + MaxBlockFetch
+		}
+		return persistence.latestHeight
+	}
+
+	for i := persistence.latestQueriedBlock + 1; i <= syncUpHeight(); i++ {
 		var eg errgroup.Group
 		var blockRes *ctypes.ResultBlockResults
 		var lightBlock *types.LightBlock
@@ -434,7 +442,7 @@
 		}
 
 		if err := ccp.Verify(ctx, lightBlock); err != nil {
 			ccp.log.Warn("Failed to verify block", zap.Int64("height", blockRes.Height), zap.Error(err))
 			return err
 		}
 
@@ -460,7 +468,7 @@
 			messages := ibcMessagesFromEvents(ccp.log, tx.Events, chainID, heightUint64, ccp.chainProvider.PCfg.IbcHandlerAddress, base64Encoded)
 
 			for _, m := range messages {
-				ccp.log.Info("Detected eventlog", zap.String("eventlog", m.eventType))
+				ccp.log.Info("Detected eventlog", zap.String("eventlog", m.eventType), zap.Uint64("height", heightUint64))
 				ccp.handleMessage(ctx, m, ibcMessagesCache)
 			}
 		}
@@ -510,20 +518,20 @@
 	return nil
 }
 
 func (ccp *WasmChainProcessor) getHeightToSave(height int64) int {
 	retryAfter := ccp.Provider().ProviderConfig().GetFirstRetryBlockAfter()
 	ht := int(height - int64(retryAfter))
 	if ht < 0 {
 		return 0
 	}
 	return ht
 }
 
 func (ccp *WasmChainProcessor) SnapshotHeight(height int) {
 	ccp.log.Info("Save height for snapshot", zap.Int("height", height))
 	err := common.SnapshotHeight(ccp.Provider().ChainId(), height)
 	if err != nil {
 		ccp.log.Warn("Failed saving height snapshot for height", zap.Int("height", height))
 	}
 }
 

relayer/common/const.go

@@ -15,7 +15,7 @@ var (
 	ConnectionKey         = "connection"
 	ChannelKey            = "channel"
 	ONE_HOUR              = 60 * 60 * 1000
-	NanosecondRatio       = 1000_000
+	NanoToMilliRatio      = 1000_000
 )
 
 var (