feat: log height every hour simulateneously

relayer/chains/icon/icon_chain_processor.go

@@ -66,6 +66,8 @@
 	metrics *processor.PrometheusMetrics
 
 	verifier *Verifier
+
+	heightSnapshotChan chan struct{}
 }
 
 type Verifier struct {
@@ -74,9 +76,9 @@
 	prevNetworkSectionHash []byte
 }
 
-func NewIconChainProcessor(log *zap.Logger, provider *IconProvider, metrics *processor.PrometheusMetrics) *IconChainProcessor {
+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),
@@ -84,6 +86,7 @@
 		connectionClients:    make(map[string]string),
 		channelConnections:   make(map[string]string),
 		metrics:              metrics,
+		heightSnapshotChan:   heightSnapshot,
 	}
 }
 
@@ -149,7 +152,7 @@
 	}
 
 	// start_query_cycle
 	icp.log.Debug("Starting query cycle")
 	err := icp.monitoring(ctx, &persistence)
 	return err
 }
@@ -168,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 {
@@ -195,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
@@ -229,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
@@ -284,7 +287,7 @@
 	}
 	// }
 
 	icp.log.Info("Start to query from height", zap.Int64("height", processedheight))
 	// subscribe to monitor block
 	ctxMonitorBlock, cancelMonitorBlock := context.WithCancel(ctx)
 	reconnect()
@@ -306,13 +309,16 @@
 		case err := <-errCh:
 			return err
 
+		case <-icp.heightSnapshotChan:
+			icp.SnapshotHeight(icp.getHeightToSave(int64(icp.latestBlock.Height)))
+
 		case <-reconnectCh:
 			cancelMonitorBlock()
 			ctxMonitorBlock, cancelMonitorBlock = context.WithCancel(ctx)
 
 			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
@@ -321,10 +327,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
 					}
@@ -340,8 +346,8 @@
 				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
@@ -359,7 +365,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 {
@@ -374,10 +380,6 @@
 				if br = nil; len(btpBlockRespCh) > 0 {
 					br = <-btpBlockRespCh
 				}
-				ht, takeSnapshot := icp.shouldSnapshot(int(icp.latestBlock.Height))
-				if takeSnapshot {
-					icp.SnapshotHeight(ht)
-				}
 			}
 			// remove unprocessed blockResponses
 			for len(btpBlockRespCh) > 0 {
@@ -468,32 +470,20 @@
 	}
 }
 
-func (icp *IconChainProcessor) shouldSnapshot(height int) (int, bool) {
-	blockInterval := icp.Provider().ProviderConfig().GetBlockInterval()
-	snapshotThreshold := rlycommon.ONE_HOUR / int(blockInterval)
-
-	snapshotHeight := icp.getHeightToSave(int64(height))
-
-	if snapshotHeight%snapshotThreshold == 0 {
-		return snapshotHeight, true
-	}
-	return 0, false
-}
-
 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))
 	}
 }
 
@@ -552,8 +542,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
 }
 
@@ -630,8 +620,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)
 				}
 
@@ -657,7 +647,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/icon/query.go

@@ -105,7 +105,7 @@
 	if block != nil {
 		return block.Height, nil
 	}
-	return 0, fmt.Errorf("failed to query Block")
+	return 0, fmt.Errorf("failed to query latest block")
 }
 
 // legacy
@@ -615,12 +615,12 @@
 				"portId":    portId,
 			}), &_channel)
 			if err != nil {
 				icp.log.Error("unable to fetch channel for  ", zap.String("channel-id ", channelId), zap.Error(err))
 				continue
 			}
 
 			if _channel == "" {
 				icp.log.Debug("Channel not present for ", zap.String("channel-id ", channelId), zap.String("port-id ", portId))
 				continue
 			}
 
@@ -628,7 +628,7 @@
 			_, err = HexBytesToProtoUnmarshal(_channel, &channel)
 			if err != nil {
 				icp.log.Info("Unable to unmarshal channel for ",
 					zap.String("channel-id ", channelId), zap.Error(err))
 				continue
 			}
 
@@ -673,24 +673,24 @@
 		"portId":    portid,
 		"channelId": channelid,
 	}, callParamsWithHeight(types.NewHexInt(height)))
 	var nextSeqRecv types.HexInt
 	if err := icp.client.Call(callParam, &nextSeqRecv); err != nil {
 		return nil, err
 	}
 	key := common.GetNextSequenceRecvCommitmentKey(portid, channelid)
 	keyHash := common.Sha3keccak256(key, []byte(nextSeqRecv))
 
 	proof, err := icp.QueryIconProof(ctx, height, keyHash)
 	if err != nil {
 		return nil, err
 	}
 
 	nextSeq, err := nextSeqRecv.Value()
 	if err != nil {
 		return nil, err
 	}
 	return &chantypes.QueryNextSequenceReceiveResponse{
 		NextSequenceReceive: uint64(nextSeq),
 		Proof:               proof,
 		ProofHeight:         clienttypes.NewHeight(0, uint64(height)),
 	}, nil

relayer/chains/wasm/wasm_chain_processor.go

@@ -58,13 +58,15 @@
 	parsedGasPrices *sdk.DecCoins
 
 	verifier *Verifier
+
+	heightSnapshotChan chan struct{}
 }
 
 type Verifier struct {
 	Header *types.LightBlock
 }
 
-func NewWasmChainProcessor(log *zap.Logger, provider *WasmProvider, metrics *processor.PrometheusMetrics) *WasmChainProcessor {
+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,
@@ -74,6 +76,7 @@
 		connectionClients:    make(map[string]string),
 		channelConnections:   make(map[string]string),
 		metrics:              metrics,
+		heightSnapshotChan:   heightSnapshot,
 	}
 }
 
@@ -218,7 +221,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)),
 		lastBalanceUpdate:         time.Unix(0, 0),
 		balanceUpdateWaitDuration: defaultBalanceUpdateWaitDuration,
 	}
@@ -290,6 +293,8 @@
 		select {
 		case <-ctx.Done():
 			return nil
+		case <-ccp.heightSnapshotChan:
+			ccp.SnapshotHeight(ccp.getHeightToSave(persistence.latestHeight))
 		case <-ticker.C:
 			ticker.Reset(persistence.minQueryLoopDuration)
 		}
@@ -312,11 +317,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
 }
@@ -345,11 +350,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
 }
@@ -363,15 +368,16 @@
 			zap.Uint("attempts", latestHeightQueryRetries),
 			zap.Error(err),
 		)
 
-		ccp.SnapshotHeight(ccp.getHeightToSave(status.SyncInfo.LatestBlockHeight))
+		// 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),
 	)
 
@@ -404,11 +410,6 @@
 	chainID := ccp.chainProvider.ChainId()
 	var latestHeader provider.IBCHeader
 
-	ht, takeSnapshot := ccp.shouldSnapshot(int(persistence.latestHeight))
-	if takeSnapshot {
-		ccp.SnapshotHeight(ht)
-	}
-
 	for i := persistence.latestQueriedBlock + 1; i <= persistence.latestHeight; i++ {
 		var eg errgroup.Group
 		var blockRes *ctypes.ResultBlockResults
@@ -433,7 +434,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
 		}
 
@@ -459,7 +460,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.handleMessage(ctx, m, ibcMessagesCache)
 			}
 		}
@@ -509,32 +510,20 @@
 	return nil
 }
 
-func (ccp *WasmChainProcessor) shouldSnapshot(height int) (int, bool) {
-	blockInterval := ccp.Provider().ProviderConfig().GetBlockInterval()
-	snapshotThreshold := common.ONE_HOUR / int(blockInterval)
-
-	snapshotHeight := ccp.getHeightToSave(int64(height))
-
-	if snapshotHeight%snapshotThreshold == 0 {
-		return snapshotHeight, true
-	}
-	return 0, false
-}
-
 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/channel.go

@@ -71,8 +71,8 @@
 
 	return processor.NewEventProcessor().
 		WithChainProcessors(
-			c.chainProcessor(c.log, nil),
-			dst.chainProcessor(c.log, nil),
+			c.chainProcessor(c.log, nil, nil),
+			dst.chainProcessor(c.log, nil, nil),
 		).
 		WithPathProcessors(pp).
 		WithInitialBlockHistory(0).
@@ -121,8 +121,8 @@
 
 	flushProcessor := processor.NewEventProcessor().
 		WithChainProcessors(
-			c.chainProcessor(c.log, nil),
-			dst.chainProcessor(c.log, nil),
+			c.chainProcessor(c.log, nil, nil),
+			dst.chainProcessor(c.log, nil, nil),
 		).
 		WithPathProcessors(processor.NewPathProcessor(
 			c.log,
@@ -159,8 +159,8 @@
 
 	return processor.NewEventProcessor().
 		WithChainProcessors(
-			c.chainProcessor(c.log, nil),
-			dst.chainProcessor(c.log, nil),
+			c.chainProcessor(c.log, nil, nil),
+			dst.chainProcessor(c.log, nil, nil),
 		).
 		WithPathProcessors(processor.NewPathProcessor(
 			c.log,

relayer/connection.go

@@ -61,8 +61,8 @@
 
 	return connectionSrc, connectionDst, processor.NewEventProcessor().
 		WithChainProcessors(
-			c.chainProcessor(c.log, nil),
-			dst.chainProcessor(c.log, nil),
+			c.chainProcessor(c.log, nil, nil),
+			dst.chainProcessor(c.log, nil, nil),
 		).
 		WithPathProcessors(pp).
 		WithInitialBlockHistory(initialBlockHistory).

relayer/strategies.go

@@ -33,6 +33,27 @@
 	TwoMB                               = 2 * 1024 * 1024
 )
 
+func timerChannel(ctx context.Context, log *zap.Logger, timerChan map[string]chan struct{}, chains map[string]*Chain) {
+	ticker := time.NewTicker(time.Hour)
+	defer ticker.Stop()
+	for {
+	NamedLoop:
+		select {
+		case <-ticker.C:
+			for _, c := range chains {
+				_, err := c.ChainProvider.QueryLatestHeight(ctx)
+				if err != nil {
+					log.Warn("Failed getting status of chain", zap.String("chain_id", c.ChainID()), zap.Error(err))
+					break NamedLoop
+				}
+			}
+			for _, c := range timerChan {
+				c <- struct{}{}
+			}
+		}
+	}
+}
+
 // StartRelayer starts the main relaying loop and returns a channel that will contain any control-flow related errors.
 func StartRelayer(
 	ctx context.Context,
@@ -49,13 +70,20 @@
 	metrics *processor.PrometheusMetrics,
 ) chan error {
 	errorChan := make(chan error, 1)
+	chans := make(map[string]chan struct{})
+
+	for k := range chains {
+		chans[k] = make(chan struct{})
+	}
+
+	go timerChannel(ctx, log, chans, chains)
 
 	switch processorType {
 	case ProcessorEvents:
 		chainProcessors := make([]processor.ChainProcessor, 0, len(chains))
 
-		for _, chain := range chains {
-			chainProcessors = append(chainProcessors, chain.chainProcessor(log, metrics))
+		for name, chain := range chains {
+			chainProcessors = append(chainProcessors, chain.chainProcessor(log, metrics, chans[name]))
 		}
 
 		ePaths := make([]path, len(paths))
@@ -116,17 +144,17 @@
 }
 
 // chainProcessor returns the corresponding ChainProcessor implementation instance for a pathChain.
-func (chain *Chain) chainProcessor(log *zap.Logger, metrics *processor.PrometheusMetrics) processor.ChainProcessor {
+func (chain *Chain) chainProcessor(log *zap.Logger, metrics *processor.PrometheusMetrics, timerChan chan struct{}) processor.ChainProcessor {
 	// Handle new ChainProcessor implementations as cases here
 	switch p := chain.ChainProvider.(type) {
 	case *penumbraprocessor.PenumbraProvider:
 		return penumbraprocessor.NewPenumbraChainProcessor(log, p)
 	case *cosmos.CosmosProvider:
 		return cosmos.NewCosmosChainProcessor(log, p, metrics)
 	case *icon.IconProvider:
-		return icon.NewIconChainProcessor(log, p, metrics)
+		return icon.NewIconChainProcessor(log, p, metrics, timerChan)
 	case *wasm.WasmProvider:
-		return wasm.NewWasmChainProcessor(log, p, metrics)
+		return wasm.NewWasmChainProcessor(log, p, metrics, timerChan)
 	default:
 		panic(fmt.Errorf("unsupported chain provider type: %T", chain.ChainProvider))
 	}