fix: Observed Packet Metric (#1345)

* fix: restructure `ShouldRelayChannel`

* fix test

* check channel open + tests

* remove open channel check

* fix race - remove warn log

* add mutex's

* Update comment

interchaintest/go.mod

@@ -172,7 +172,6 @@ require (
 	github.com/manifoldco/promptui v0.9.0 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
-	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
 	github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0 // indirect
 	github.com/mimoo/StrobeGo v0.0.0-20220103164710-9a04d6ca976b // indirect
 	github.com/minio/highwayhash v1.0.2 // indirect
@@ -213,13 +212,15 @@ require (
 	github.com/rogpeppe/go-internal v1.11.0 // indirect
 	github.com/rs/cors v1.8.3 // indirect
 	github.com/rs/zerolog v1.31.0 // indirect
+	github.com/sagikazarmark/locafero v0.3.0 // indirect
+	github.com/sagikazarmark/slog-shim v0.1.0 // indirect
 	github.com/sasha-s/go-deadlock v0.3.1 // indirect
 	github.com/sirupsen/logrus v1.9.3 // indirect
+	github.com/sourcegraph/conc v0.3.0 // indirect
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
 	github.com/spf13/afero v1.10.0 // indirect
 	github.com/spf13/cast v1.5.1 // indirect
 	github.com/spf13/cobra v1.8.0 // indirect
-	github.com/spf13/jwalterweatherman v1.1.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/spf13/viper v1.17.0 // indirect
 	github.com/strangelove-ventures/cometbft v0.37.3-0.20231004194858-c01e8d5bcac3 // indirect

relayer/processor/path_end.go

@@ -67,27 +67,3 @@ func (pe PathEnd) shouldRelayChannelSingle(channelKey ChainChannelKey, listChann
 	}
 	return !allowList
 }
-
-// if port ID is empty on allowlist channel, allow all ports
-// if port ID is non-empty on allowlist channel, allow only that specific port
-// if port ID is empty on blocklist channel, block all ports
-// if port ID is non-empty on blocklist channel, block only that specific port
-func (pe PathEnd) ShouldRelayChannel(channelKey ChainChannelKey) bool {
-	if pe.Rule == RuleAllowList {
-		for _, allowedChannel := range pe.FilterList {
-			if pe.shouldRelayChannelSingle(channelKey, allowedChannel, true) {
-				return true
-			}
-		}
-		return false
-	} else if pe.Rule == RuleDenyList {
-		for _, blockedChannel := range pe.FilterList {
-			if !pe.shouldRelayChannelSingle(channelKey, blockedChannel, false) {
-				return false
-			}
-		}
-		return true
-	}
-	// if neither allow list or block list are provided, all channels are okay
-	return true
-}

relayer/processor/path_end_runtime.go

@@ -30,6 +30,7 @@ type pathEndRuntime struct {
 	clientTrustedState   provider.ClientTrustedState
 	connectionStateCache ConnectionStateCache
 	channelStateCache    ChannelStateCache
+	channelStateCacheMu  sync.RWMutex
 	channelOrderCache    map[string]chantypes.Order
 	latestHeader         provider.IBCHeader
 	ibcHeaderCache       IBCHeaderCache
@@ -107,7 +108,7 @@ func (pathEnd *pathEndRuntime) mergeMessageCache(messageCache IBCMessagesCache,
 	clientICQMessages := make(ClientICQMessagesCache)
 
 	for ch, pmc := range messageCache.PacketFlow {
-		if pathEnd.info.ShouldRelayChannel(ChainChannelKey{ChainID: pathEnd.info.ChainID, CounterpartyChainID: counterpartyChainID, ChannelKey: ch}) {
+		if pathEnd.ShouldRelayChannel(ChainChannelKey{ChainID: pathEnd.info.ChainID, CounterpartyChainID: counterpartyChainID, ChannelKey: ch}) {
 			if inSync && pathEnd.metrics != nil {
 				for eventType, pCache := range pmc {
 					pathEnd.metrics.AddPacketsObserved(pathEnd.info.PathName, pathEnd.info.ChainID, ch.ChannelID, ch.PortID, eventType, len(pCache))
@@ -403,7 +404,10 @@ func (pathEnd *pathEndRuntime) mergeCacheData(ctx context.Context, cancel func()
 	}
 
 	pathEnd.connectionStateCache = d.ConnectionStateCache // Update latest connection open state for chain
-	pathEnd.channelStateCache = d.ChannelStateCache       // Update latest channel open state for chain
+
+	pathEnd.channelStateCacheMu.Lock()
+	pathEnd.channelStateCache = d.ChannelStateCache // Update latest channel open state for chain
+	pathEnd.channelStateCacheMu.Unlock()
 
 	pathEnd.mergeMessageCache(d.IBCMessagesCache, counterpartyChainID, pathEnd.inSync && counterpartyInSync) // Merge incoming packet IBC messages into the backlog
 
@@ -874,3 +878,44 @@ func (pathEnd *pathEndRuntime) localhostSentinelProofChannel(
 		Version:     info.Version,
 	}, nil
 }
+
+// ShouldRelayChannel determines whether a chain channel (and optionally a port), should be relayed
+// by this path end.
+//
+// It first checks if the channel matches any rule in the path end's filter list. If the channel matches a channel
+// in an allowed list, it returns true. If the channel matches any blocked channel it returns false. Otherwise, it returns true.
+//
+// If no filter rule matches, it checks if the channel or its counterparty is present in the path end's
+// channel state cache. This cache only holds channels relevant to the client for this path end, ensuring
+// the channel belongs to a client connected to this path end.
+//
+// Note that this function only determines whether the channel should be relayed based on the path end's
+// configuration. It does not guarantee that the channel is actually relayable, as other checks
+// (e.g., expired client) may still be necessary.
+func (pathEnd *pathEndRuntime) ShouldRelayChannel(chainChannelKey ChainChannelKey) bool {
+	pe := pathEnd.info
+	if pe.Rule == RuleAllowList {
+		for _, allowedChannel := range pe.FilterList {
+			if pe.shouldRelayChannelSingle(chainChannelKey, allowedChannel, true) {
+				return true
+			}
+		}
+		return false
+	} else if pe.Rule == RuleDenyList {
+		for _, blockedChannel := range pe.FilterList {
+			if !pe.shouldRelayChannelSingle(chainChannelKey, blockedChannel, false) {
+				return false
+			}
+		}
+		return true
+	}
+
+	pathEnd.channelStateCacheMu.RLock()
+	defer pathEnd.channelStateCacheMu.RUnlock()
+
+	// if no filter rule, check if the channel or counterparty channel is in the channelStateCache.
+	// Because channelStateCache only holds channels relevant to the client, we can ensure that the
+	// channel is built on top of a client for this pathEnd
+	_, exists := pathEnd.channelStateCache[chainChannelKey.ChannelKey]
+	return exists
+}