fix: fix start height greater than latest height issue

relayer/chains/cosmos/cosmos_chain_processor.go

@@ -262,11 +262,11 @@
 	}
 }
 
-func (ccp *CosmosChainProcessor) SnapshotHeight(height int) {
+func (ccp *CosmosChainProcessor) SnapshotHeight(height int64) {
 	panic("Not implemented for Cosmos")
 }
 
-func (ccp *CosmosChainProcessor) StartFromHeight(ctx context.Context) int {
+func (ccp *CosmosChainProcessor) StartFromHeight(ctx context.Context) int64 {
 	panic("Not implemented for Cosmos")
 }
 

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,29 +152,30 @@
 	}
 
 	// start_query_cycle
 	icp.log.Debug("Starting query cycle")
 	err := icp.monitoring(ctx, &persistence)
 	return err
 }
 
-func (icp *IconChainProcessor) StartFromHeight(ctx context.Context) int {
+func (icp *IconChainProcessor) StartFromHeight(ctx context.Context) int64 {
 	cfg := icp.Provider().ProviderConfig().(*IconProviderConfig)
+
 	if cfg.StartHeight != 0 {
-		return int(cfg.StartHeight)
+		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.Int("height", snapshotHeight))
+		icp.log.Info("Obtained start height from config", zap.Int64("height", snapshotHeight))
 	}
 	return snapshotHeight
 }
 
-func (icp *IconChainProcessor) getLastSavedHeight() int {
+func (icp *IconChainProcessor) getLastSavedHeight() int64 {
 	snapshotHeight, err := rlycommon.LoadSnapshotHeight(icp.Provider().ChainId())
 	if err != nil || snapshotHeight < 0 {
 		return 0
 	}
 	return snapshotHeight
 }
@@ -198,9 +199,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 +233,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
@@ -275,18 +276,24 @@
 	}
 
 	var err error
-	// processedheight := int64(icp.chainProvider.lastBTPBlockHeight)
-	// if processedheight == 0 {
-	processedheight := int64(icp.StartFromHeight(ctx))
+	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, err = icp.chainProvider.QueryLatestHeight(ctx)
-		if err != nil {
-			fmt.Println("Error fetching latest block")
-			return err
-		}
+		processedheight = latestHeight
 	}
 
 	icp.log.Info("Start to query from height", zap.Int64("height", processedheight))
 	// subscribe to monitor block
 	ctxMonitorBlock, cancelMonitorBlock := context.WithCancel(ctx)
 	reconnect()
@@ -296,7 +303,7 @@
 	icp.firstTime = true
 
 	blockReq := &types.BlockRequest{
 		Height:       types.NewHexInt(int64(processedheight)),
 		EventFilters: GetMonitorEventFilters(icp.chainProvider.PCfg.IbcHandlerAddress),
 	}
 
@@ -308,8 +315,8 @@
 		case err := <-errCh:
 			return err
 
 		case <-icp.heightSnapshotChan:
 			icp.SnapshotHeight(icp.getHeightToSave(int64(icp.latestBlock.Height)))
 
 		case <-reconnectCh:
 			cancelMonitorBlock()
@@ -317,7 +324,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
@@ -326,10 +333,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
 					}
@@ -345,16 +352,16 @@
 				err := icp.verifyBlock(ctx, br.Header)
 				if err != nil {
 					reconnect()
-					icp.log.Warn("Failed to verify BTP Block",
+					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),
 				}
@@ -367,7 +374,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 {
@@ -378,9 +385,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
@@ -402,7 +409,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),
 						)
@@ -475,20 +482,20 @@
 	}
 }
 
-func (icp *IconChainProcessor) getHeightToSave(height int64) int {
+func (icp *IconChainProcessor) getHeightToSave(height int64) int64 {
 	retryAfter := icp.Provider().ProviderConfig().GetFirstRetryBlockAfter()
-	ht := int(height - int64(retryAfter))
+	ht := 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))
+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.Int("height", height))
+		icp.log.Warn("Failed saving height snapshot for height", zap.Int64("height", height))
 	}
 }
 
@@ -547,8 +554,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
 }
 
@@ -625,8 +632,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)
 				}
 
@@ -652,7 +659,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/mock/mock_chain_processor.go

@@ -59,11 +59,11 @@ func NewMockChainProcessor(ctx context.Context, log *zap.Logger, chainID string,
 	}
 }
 
-func (mcp *MockChainProcessor) SnapshotHeight(height int) {
+func (mcp *MockChainProcessor) SnapshotHeight(height int64) {
 	panic("")
 }
 
-func (mcp *MockChainProcessor) StartFromHeight(ctx context.Context) int {
+func (mcp *MockChainProcessor) StartFromHeight(ctx context.Context) int64 {
 	return 0
 }
 

relayer/chains/penumbra/penumbra_chain_processor.go

@@ -267,11 +267,11 @@ func (pcp *PenumbraChainProcessor) initializeChannelState(ctx context.Context) e
 	return nil
 }
 
-func (ccp *PenumbraChainProcessor) SnapshotHeight(height int) {
+func (ccp *PenumbraChainProcessor) SnapshotHeight(height int64) {
 	panic("Not implemented for Penumbra")
 }
 
-func (ccp *PenumbraChainProcessor) StartFromHeight(ctx context.Context) int {
+func (ccp *PenumbraChainProcessor) StartFromHeight(ctx context.Context) int64 {
 	panic("Not implemented for Penumbra")
 }
 

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) int {
+func (ccp *WasmChainProcessor) StartFromHeight(ctx context.Context) int64 {
 	cfg := ccp.Provider().ProviderConfig().(*WasmProviderConfig)
 	if cfg.StartHeight != 0 {
-		return int(cfg.StartHeight)
+		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.Int("height", snapshotHeight))
+		ccp.log.Info("Obtained start height from config", zap.Int64("height", snapshotHeight))
 	}
 	return snapshotHeight
 }
@@ -223,7 +223,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.NanoToMilliRatio)),
 		lastBalanceUpdate:         time.Unix(0, 0),
 		balanceUpdateWaitDuration: defaultBalanceUpdateWaitDuration,
 	}
@@ -248,16 +248,13 @@
 
 	// this will make initial QueryLoop iteration look back initialBlockHistory blocks in history
 	latestQueriedBlock := ccp.StartFromHeight(ctx)
-	if latestQueriedBlock < 0 {
-		latestQueriedBlock = int(persistence.latestHeight - int64(initialBlockHistory))
-		if latestQueriedBlock < 0 {
-			latestQueriedBlock = 0
-		}
+	if latestQueriedBlock <= 0 || latestQueriedBlock > persistence.latestHeight {
+		latestQueriedBlock = persistence.latestHeight
 	}
 
 	persistence.latestQueriedBlock = int64(latestQueriedBlock)
 
-	ccp.log.Info("Start to query from height ", zap.Int("height", latestQueriedBlock))
+	ccp.log.Info("Start to query from height ", zap.Int64("height", latestQueriedBlock))
 
 	_, lightBlock, err := ccp.chainProvider.QueryLightBlock(ctx, persistence.latestQueriedBlock)
 	if err != nil {
@@ -295,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)
 		}
@@ -319,11 +316,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
 }
@@ -352,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
 }
@@ -370,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),
 	)
 
@@ -412,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
@@ -443,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
 		}
 
@@ -469,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)
 			}
 		}
@@ -519,20 +516,20 @@
 	return nil
 }
 
-func (ccp *WasmChainProcessor) getHeightToSave(height int64) int {
+func (ccp *WasmChainProcessor) getHeightToSave(height int64) int64 {
 	retryAfter := ccp.Provider().ProviderConfig().GetFirstRetryBlockAfter()
-	ht := int(height - int64(retryAfter))
+	ht := 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))
+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.Int("height", height))
+		ccp.log.Warn("Failed saving height snapshot for height", zap.Int64("height", height))
 	}
 }
 

relayer/common/utils.go

@@ -29,7 +29,7 @@ func getSnapshotPath(chain_name string) (string, error) {
 	return snapshot, nil
 }
 
-func SnapshotHeight(chain_id string, height int) error {
+func SnapshotHeight(chain_id string, height int64) error {
 	snapshot, err := getSnapshotPath(chain_id)
 	if err != nil {
 		return fmt.Errorf("Failed to find snapshot path, %w", err)
@@ -46,7 +46,7 @@ func SnapshotHeight(chain_id string, height int) error {
 	return nil
 }
 
-func LoadSnapshotHeight(chain_id string) (int, error) {
+func LoadSnapshotHeight(chain_id string) (int64, error) {
 	snapshot, err := getSnapshotPath(chain_id)
 	if err != nil {
 		return -1, fmt.Errorf("Failed to find snapshot path, %w", err)
@@ -56,5 +56,5 @@ func LoadSnapshotHeight(chain_id string) (int, error) {
 	if err != nil {
 		return -1, fmt.Errorf("Failed reading file, %w", err)
 	}
-	return strconv.Atoi(strings.TrimSuffix(string(content), "\n"))
+	return strconv.ParseInt(strings.TrimSuffix(string(content), "\n"), 10, 64)
 }

relayer/processor/chain_processor.go

@@ -23,11 +23,11 @@ type ChainProcessor interface {
 
 	// Take snapshot of height every N blocks or when the chain processor fails, so that the relayer
 	// can restart from that height
-	SnapshotHeight(height int)
+	SnapshotHeight(height int64)
 
 	// If the relay goes down, start chain processor from height returned by this function
 	// CAN return max(snapshotHeight, latestHeightFromClient)
-	StartFromHeight(ctx context.Context) int
+	StartFromHeight(ctx context.Context) int64
 }
 
 // ChainProcessors is a slice of ChainProcessor instances.