fix: remove context timeout from initalize connection/channel state

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,38 +152,37 @@
 	}
 
 	// start_query_cycle
 	icp.log.Debug("Starting query cycle")
 	err := icp.monitoring(ctx, &persistence)
 	return err
 }
 
 func (icp *IconChainProcessor) StartFromHeight(ctx context.Context) int64 {
 	cfg := icp.Provider().ProviderConfig().(*IconProviderConfig)
 
 	if cfg.StartHeight != 0 {
 		return cfg.StartHeight
 	}
 	snapshotHeight, err := rlycommon.LoadSnapshotHeight(icp.Provider().ChainId())
 	if err != nil {
 		icp.log.Warn("Failed to load height from snapshot", zap.Error(err))
 	} else {
 		icp.log.Info("Obtained start height from config", zap.Int64("height", snapshotHeight))
 	}
 	return snapshotHeight
 }
 
 func (icp *IconChainProcessor) getLastSavedHeight() int64 {
 	snapshotHeight, err := rlycommon.LoadSnapshotHeight(icp.Provider().ChainId())
 	if err != nil || snapshotHeight < 0 {
 		return 0
 	}
 	return snapshotHeight
 }
 
 func (icp *IconChainProcessor) initializeConnectionState(ctx context.Context) error {
-	// TODO: review
-	ctx, cancel := context.WithTimeout(ctx, queryTimeout)
-	defer cancel()
+	// ctx, cancel := context.WithTimeout(ctx, queryTimeout)
+	// defer cancel()
 
 	connections, err := icp.chainProvider.QueryConnections(ctx)
 	if err != nil {
@@ -199,18 +198,17 @@
 			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
 }
 
 func (icp *IconChainProcessor) initializeChannelState(ctx context.Context) error {
-	// TODO:
-	ctx, cancel := context.WithTimeout(ctx, queryTimeout)
-	defer cancel()
+	// ctx, cancel := context.WithTimeout(ctx, queryTimeout)
+	// defer cancel()
 	channels, err := icp.chainProvider.QueryChannels(ctx)
 	if err != nil {
 		return fmt.Errorf("error querying channels: %w", err)
@@ -233,11 +231,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
@@ -276,24 +274,24 @@
 	}
 
 	var err error
 	processedheight := icp.StartFromHeight(ctx)
 	latestHeight, err := icp.chainProvider.QueryLatestHeight(ctx)
 	if err != nil {
 		icp.log.Error("Error fetching block", zap.Error(err))
 		return err
 	}
 	if processedheight > latestHeight {
 		icp.log.Warn("Start height set is greater than latest height",
 			zap.Int64("start height", processedheight),
 			zap.Int64("latest Height", latestHeight),
 		)
 		processedheight = latestHeight
 	}
 	if processedheight <= 0 {
 		processedheight = latestHeight
 	}
 
 	icp.log.Info("Start to query from height", zap.Int64("height", processedheight))
 	// subscribe to monitor block
 	ctxMonitorBlock, cancelMonitorBlock := context.WithCancel(ctx)
 	reconnect()
@@ -303,7 +301,7 @@
 	icp.firstTime = true
 
 	blockReq := &types.BlockRequest{
 		Height:       types.NewHexInt(int64(processedheight)),
 		EventFilters: GetMonitorEventFilters(icp.chainProvider.PCfg.IbcHandlerAddress),
 	}
 
@@ -315,8 +313,8 @@
 		case err := <-errCh:
 			return err
 
 		case <-icp.heightSnapshotChan:
 			icp.SnapshotHeight(icp.getHeightToSave(int64(icp.latestBlock.Height)))
 
 		case <-reconnectCh:
 			cancelMonitorBlock()
@@ -324,7 +322,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
@@ -333,10 +331,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
 					}
@@ -352,16 +350,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),
 				}
@@ -374,7 +372,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 {
@@ -385,9 +383,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
@@ -409,7 +407,7 @@
 					if err != nil {
 						return err
 					} else if height != processedheight+i {
 						icp.log.Warn("Reconnect: missing block notification",
 							zap.Int64("got", height),
 							zap.Int64("expected", processedheight+i),
 						)
@@ -482,20 +480,20 @@
 	}
 }
 
 func (icp *IconChainProcessor) getHeightToSave(height int64) int64 {
 	retryAfter := icp.Provider().ProviderConfig().GetFirstRetryBlockAfter()
 	ht := height - int64(retryAfter)
 	if ht < 0 {
 		return 0
 	}
 	return ht
 }
 
 func (icp *IconChainProcessor) SnapshotHeight(height int64) {
 	icp.log.Info("Save height for snapshot", zap.Int64("height", height))
 	err := rlycommon.SnapshotHeight(icp.Provider().ChainId(), height)
 	if err != nil {
 		icp.log.Warn("Failed saving height snapshot for height", zap.Int64("height", height))
 	}
 }
 
@@ -554,8 +552,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
 }
 
@@ -632,8 +630,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)
 				}
 
@@ -659,7 +657,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))

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,
 	}
 }
 
@@ -178,7 +178,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 {
 		return state, nil
 	}
 	cs, err := ccp.chainProvider.QueryClientState(ctx, int64(ccp.latestBlock.Height), clientID)
@@ -203,16 +203,16 @@
 	balanceUpdateWaitDuration time.Duration
 }
 
 func (ccp *WasmChainProcessor) StartFromHeight(ctx context.Context) int64 {
 	cfg := ccp.Provider().ProviderConfig().(*WasmProviderConfig)
 	if cfg.StartHeight != 0 {
 		return int64(cfg.StartHeight)
 	}
 	snapshotHeight, err := common.LoadSnapshotHeight(ccp.Provider().ChainId())
 	if err != nil {
 		ccp.log.Warn("Failed to load height from snapshot", zap.Error(err))
 	} else {
 		ccp.log.Info("Obtained start height from config", zap.Int64("height", snapshotHeight))
 	}
 	return snapshotHeight
 }
@@ -248,13 +248,13 @@
 
 	// this will make initial QueryLoop iteration look back initialBlockHistory blocks in history
 	latestQueriedBlock := ccp.StartFromHeight(ctx)
 	if latestQueriedBlock <= 0 || latestQueriedBlock > persistence.latestHeight {
 		latestQueriedBlock = persistence.latestHeight
 	}
 
 	persistence.latestQueriedBlock = int64(latestQueriedBlock)
 
 	ccp.log.Info("Start to query from height ", zap.Int64("height", latestQueriedBlock))
 
 	_, lightBlock, err := ccp.chainProvider.QueryLightBlock(ctx, persistence.latestQueriedBlock)
 	if err != nil {
@@ -292,8 +292,8 @@
 		select {
 		case <-ctx.Done():
 			return nil
 		case <-ccp.heightSnapshotChan:
 			ccp.SnapshotHeight(ccp.getHeightToSave(persistence.latestHeight))
 		case <-ticker.C:
 			ticker.Reset(persistence.minQueryLoopDuration)
 		}
@@ -302,8 +302,8 @@
 
 // initializeConnectionState will bootstrap the connectionStateCache with the open connection state.
 func (ccp *WasmChainProcessor) initializeConnectionState(ctx context.Context) error {
-	ctx, cancel := context.WithTimeout(ctx, queryTimeout)
-	defer cancel()
+	// ctx, cancel := context.WithTimeout(ctx, queryTimeout)
+	// defer cancel()
 	connections, err := ccp.chainProvider.QueryConnections(ctx)
 	if err != nil {
 		return fmt.Errorf("error querying connections: %w", err)
@@ -316,19 +316,19 @@
 			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
 }
 
 // initializeChannelState will bootstrap the channelStateCache with the open channel state.
 func (ccp *WasmChainProcessor) initializeChannelState(ctx context.Context) error {
-	ctx, cancel := context.WithTimeout(ctx, queryTimeout)
-	defer cancel()
+	// ctx, cancel := context.WithTimeout(ctx, queryTimeout)
+	// defer cancel()
 	channels, err := ccp.chainProvider.QueryChannels(ctx)
 	if err != nil {
 		return fmt.Errorf("error querying channels: %w", err)
@@ -349,11 +349,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
 }
@@ -367,16 +367,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),
 	)
 
@@ -409,14 +409,14 @@
 	chainID := ccp.chainProvider.ChainId()
 	var latestHeader provider.IBCHeader
 
 	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
@@ -440,7 +440,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
 		}
 
@@ -466,7 +466,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), zap.Uint64("height", heightUint64))
 				ccp.handleMessage(ctx, m, ibcMessagesCache)
 			}
 		}
@@ -516,20 +516,20 @@
 	return nil
 }
 
 func (ccp *WasmChainProcessor) getHeightToSave(height int64) int64 {
 	retryAfter := ccp.Provider().ProviderConfig().GetFirstRetryBlockAfter()
 	ht := height - int64(retryAfter)
 	if ht < 0 {
 		return 0
 	}
 	return ht
 }
 
 func (ccp *WasmChainProcessor) SnapshotHeight(height int64) {
 	ccp.log.Info("Save height for snapshot", zap.Int64("height", height))
 	err := common.SnapshotHeight(ccp.Provider().ChainId(), height)
 	if err != nil {
 		ccp.log.Warn("Failed saving height snapshot for height", zap.Int64("height", height))
 	}
 }