Repository: cosmos/relayer
Pull request #777 (2022-06-07 02:25:08)

CosmosChainProcessor - IBC message handlers
IBC message handlers for caching data and passing to path processors, similar to the ones mocked in `MockChainProcessor`:
- `MsgTransfer`: tracks message so that `PathProcessor` can handle it, additionally constructs information that will be needed for `MsgRecvPacket` on the counterparty chain.
- `MsgRecvPacket`: tracks message so that `PathProcessor` can handle it, constructs information that will be needed for `MsgAcknowledgement` on the counterparty chain.
- `MsgAcknowledgement`, `MsgTimeout`, `MsgTimeoutOnClose`: tracks message so that `PathProcessor` can handle it. No additional information is needed since these are packet flow complete messages.

Introduces `ChannelStateCache` type for keeping channel open state in sync with `PathProcessor`s. `PathProcessor` will only handle packet messages for open channels. Additional packet filtering based on channel filter will be in future PR.

Updates `MockChainProcessor` and associated test to be channel-aware.

boojamya: Closes https://github.com/cosmos/relayer/issues/742
DavidNix: That viz is helpful, thanks!

There's a lot going on in the Chain Processor. I'm wondering if there's a patten to break up and make visible those responsibilities. But I can't see a clear path to that now. Additionally, it may be cosmos-specific in this case. 

I would enjoy @mark-rushakoff's feedback as well. 

Patch:
diff --git a/relayer/chains/cosmos/cosmos_chain_processor.go b/relayer/chains/cosmos/cosmos_chain_processor.go
index b411d3a..89e0b02 100644
--- a/relayer/chains/cosmos/cosmos_chain_processor.go
+++ b/relayer/chains/cosmos/cosmos_chain_processor.go
@@ -27,6 +27,9 @@ type CosmosChainProcessor struct {
 
 	// indicates whether queries are in sync with latest height of the chain
 	inSync bool
+
+	// holds open state for discovered channels
+	channelStateCache processor.ChannelStateCache
 }
 
 func NewCosmosChainProcessor(log *zap.Logger, provider *cosmos.CosmosProvider, rpcAddress string, input io.Reader, output io.Writer, pathProcessors processor.PathProcessors) (*CosmosChainProcessor, error) {
@@ -35,10 +38,11 @@ func NewCosmosChainProcessor(log *zap.Logger, provider *cosmos.CosmosProvider, r
 		return nil, fmt.Errorf("error getting cosmos client: %w", err)
 	}
 	return &CosmosChainProcessor{
-		log:            log.With(zap.String("chain_id", provider.ChainId())),
-		chainProvider:  provider,
-		cc:             cc,
-		pathProcessors: pathProcessors,
+		log:               log.With(zap.String("chain_id", provider.ChainId())),
+		chainProvider:     provider,
+		cc:                cc,
+		pathProcessors:    pathProcessors,
+		channelStateCache: make(processor.ChannelStateCache),
 	}, nil
 }
 
@@ -174,6 +178,10 @@ func (ccp *CosmosChainProcessor) queryCycle(ctx context.Context, persistence *qu
 		}
 	}
 
+	foundMessages := make(processor.ChannelMessageCache)
+
+	ppChanged := false
+
 	for i := persistence.latestQueriedBlock + 1; i <= persistence.latestHeight; i++ {
 		blockRes, err := ccp.cc.Client.BlockResults(ctx, &i)
 		if err != nil {
@@ -190,14 +198,40 @@ func (ccp *CosmosChainProcessor) queryCycle(ctx context.Context, persistence *qu
 
 			ccp.log.Debug("Parsed IBC messages", zap.Any("messages", messages))
 
-			// TODO pass messages to handlers
+			for _, m := range messages {
+				handler, ok := messageHandlers[m.messageType]
+				if !ok {
+					continue
+				}
+				// call message handler for this ibc message type. can do things like cache things on the chain processor or retain ibc messages that should be sent to the PathProcessors.
+				changed := handler(ccp, MsgHandlerParams{
+					messageInfo:   m.messageInfo,
+					foundMessages: foundMessages,
+				})
+				ppChanged = ppChanged || changed
+			}
 		}
 	}
 
-	if firstTimeInSync {
-		for _, pp := range ccp.pathProcessors {
-			pp.ProcessBacklogIfReady()
+	if !ppChanged {
+		if firstTimeInSync {
+			for _, pp := range ccp.pathProcessors {
+				pp.ProcessBacklogIfReady()
+			}
 		}
+
+		return nil
+	}
+
+	chainID := ccp.chainProvider.ChainId()
+
+	for _, pp := range ccp.pathProcessors {
+		pp.HandleNewData(chainID, processor.ChainProcessorCacheData{
+			ChannelMessageCache: foundMessages,
+			InSync:              ccp.inSync,
+			// fetch non-flushed channel messages. flush local cache messages, retaining Open state.
+			ChannelStateCache: ccp.channelStateCache.Flush(),
+		})
 	}
 
 	return nil
diff --git a/relayer/chains/cosmos/message_handlers.go b/relayer/chains/cosmos/message_handlers.go
new file mode 100644
index 0000000..0a53f0a
--- /dev/null
+++ b/relayer/chains/cosmos/message_handlers.go
@@ -0,0 +1,150 @@
+package cosmos
+
+import (
+	chantypes "github.com/cosmos/ibc-go/v3/modules/core/04-channel/types"
+	"github.com/cosmos/relayer/v2/relayer/processor"
+	"github.com/cosmos/relayer/v2/relayer/provider/cosmos"
+	"go.uber.org/zap"
+)
+
+type MsgHandlerParams struct {
+	messageInfo   interface{}
+	foundMessages processor.ChannelMessageCache
+}
+
+var messageHandlers = map[string]func(*CosmosChainProcessor, MsgHandlerParams) bool{
+	processor.MsgTransfer:        (*CosmosChainProcessor).handleMsgTransfer,
+	processor.MsgRecvPacket:      (*CosmosChainProcessor).handleMsgRecvPacket,
+	processor.MsgAcknowledgement: (*CosmosChainProcessor).handleMsgAcknowledgement,
+	processor.MsgTimeout:         (*CosmosChainProcessor).handleMsgTimeout,
+	processor.MsgTimeoutOnClose:  (*CosmosChainProcessor).handleMsgTimeoutOnClose,
+
+	// TODO client, connection, channel messages
+}
+
+// BEGIN packet msg handlers
+func (ccp *CosmosChainProcessor) handleMsgTransfer(p MsgHandlerParams) bool {
+	pi := p.messageInfo.(*packetInfo)
+	// source chain processor will call this handler
+	// source channel used as key because MsgTransfer is sent to source chain
+	channelKey := pi.channelKey()
+	if !p.foundMessages.ShouldRetainSequence(ccp.pathProcessors, channelKey, ccp.chainProvider.ChainId(), processor.MsgTransfer, pi.packet.Sequence) {
+		return false
+	}
+	// Construct the start of the MsgRecvPacket for the counterparty chain.
+	// PathProcessor will determine if this is needed.
+	// For example, if a MsgRecvPacket is not detected on the counterparty chain, and
+	// a MsgAcknowledgement, MsgTimeout, or MsgTimeout is not detected yet on this chain,
+	// and the packet timeout has not expired, a MsgRecvPacket will be sent to the counterparty chain
+	// using this information with the packet commitment proof from this chain added.
+	p.foundMessages.Retain(channelKey, processor.MsgTransfer, pi.packet.Sequence, cosmos.NewCosmosMessage(&chantypes.MsgRecvPacket{
+		Packet: chantypes.Packet{
+			Sequence:           pi.packet.Sequence,
+			SourcePort:         pi.packet.SourcePort,
+			SourceChannel:      pi.packet.SourceChannel,
+			DestinationPort:    pi.packet.DestinationPort,
+			DestinationChannel: pi.packet.DestinationChannel,
+			Data:               pi.packet.Data,
+			TimeoutHeight:      pi.packet.TimeoutHeight,
+			TimeoutTimestamp:   pi.packet.TimeoutTimestamp,
+		},
+	}))
+	ccp.logPacketMessage("MsgTransfer", pi,
+		zap.Uint64("timeout_height", pi.packet.TimeoutHeight.RevisionHeight),
+		zap.Uint64("timeout_height_revision", pi.packet.TimeoutHeight.RevisionNumber),
+		zap.Uint64("timeout_timestamp", pi.packet.TimeoutTimestamp),
+	)
+	return true
+}
+
+func (ccp *CosmosChainProcessor) handleMsgRecvPacket(p MsgHandlerParams) bool {
+	pi := p.messageInfo.(*packetInfo)
+	// destination chain processor will call this handler
+	// destination channel used because MsgRecvPacket is sent to destination chain
+	channelKey := pi.channelKey().Counterparty()
+	if !p.foundMessages.ShouldRetainSequence(ccp.pathProcessors, channelKey, ccp.chainProvider.ChainId(), processor.MsgRecvPacket, pi.packet.Sequence) {
+		return false
+	}
+	// Construct the start of the MsgAcknowledgement for the counterparty chain.
+	// PathProcessor will determine if this is needed.
+	// For example, if a MsgAcknowledgement is not detected yet on the counterparty chain,
+	// a MsgAcknowledgement will be sent to the counterparty chain
+	// using this information with the packet acknowledgement commitment proof from this chain added.
+	p.foundMessages.Retain(channelKey, processor.MsgRecvPacket, pi.packet.Sequence, cosmos.NewCosmosMessage(&chantypes.MsgAcknowledgement{
+		Packet: chantypes.Packet{
+			Sequence:           pi.packet.Sequence,
+			SourcePort:         pi.packet.SourcePort,
+			SourceChannel:      pi.packet.SourceChannel,
+			DestinationPort:    pi.packet.DestinationPort,
+			DestinationChannel: pi.packet.DestinationChannel,
+			Data:               pi.packet.Data,
+			TimeoutHeight:      pi.packet.TimeoutHeight,
+			TimeoutTimestamp:   pi.packet.TimeoutTimestamp,
+		},
+		Acknowledgement: pi.ack,
+	}))
+	ccp.logPacketMessage("MsgRecvPacket", pi)
+	return true
+}
+
+func (ccp *CosmosChainProcessor) handleMsgAcknowledgement(p MsgHandlerParams) bool {
+	pi := p.messageInfo.(*packetInfo)
+	// source chain processor will call this handler
+	// source channel used as key because MsgAcknowledgement is sent to source chain
+	channelKey := pi.channelKey()
+	if !p.foundMessages.ShouldRetainSequence(ccp.pathProcessors, channelKey, ccp.chainProvider.ChainId(), processor.MsgAcknowledgement, pi.packet.Sequence) {
+		return false
+	}
+	// Retaining a nil message here because this is for book-keeping in the PathProcessor cache only.
+	// A message does not need to be constructed for the counterparty chain after the MsgAcknowledgement is observed,
+	// but we want to tell the PathProcessor that the packet flow is complete for this sequence.
+	p.foundMessages.Retain(channelKey, processor.MsgAcknowledgement, pi.packet.Sequence, nil)
+	ccp.logPacketMessage("MsgAcknowledgement", pi)
+	return true
+}
+
+func (ccp *CosmosChainProcessor) handleMsgTimeout(p MsgHandlerParams) bool {
+	pi := p.messageInfo.(*packetInfo)
+	// source chain processor will call this handler
+	// source channel used as key because MsgTimeout is sent to source chain
+	channelKey := pi.channelKey()
+	if !p.foundMessages.ShouldRetainSequence(ccp.pathProcessors, channelKey, ccp.chainProvider.ChainId(), processor.MsgTimeout, pi.packet.Sequence) {
+		return false
+	}
+	// Retaining a nil message here because this is for book-keeping in the PathProcessor cache only.
+	// A message does not need to be constructed for the counterparty chain after the MsgTimeout is observed,
+	// but we want to tell the PathProcessor that the packet flow is complete for this sequence.
+	p.foundMessages.Retain(channelKey, processor.MsgTimeout, pi.packet.Sequence, nil)
+	ccp.logPacketMessage("MsgTimeout", pi)
+	return true
+}
+
+func (ccp *CosmosChainProcessor) handleMsgTimeoutOnClose(p MsgHandlerParams) bool {
+	pi := p.messageInfo.(*packetInfo)
+	// source channel used because timeout is sent to source chain
+	channelKey := pi.channelKey()
+	if !p.foundMessages.ShouldRetainSequence(ccp.pathProcessors, channelKey, ccp.chainProvider.ChainId(), processor.MsgTimeoutOnClose, pi.packet.Sequence) {
+		return false
+	}
+	// Retaining a nil message here because this is for book-keeping in the PathProcessor cache only.
+	// A message does not need to be constructed for the counterparty chain after the MsgTimeoutOnClose is observed,
+	// but we want to tell the PathProcessor that the packet flow is complete for this sequence.
+	p.foundMessages.Retain(channelKey, processor.MsgTimeoutOnClose, pi.packet.Sequence, nil)
+	ccp.logPacketMessage("MsgTimeoutOnClose", pi)
+	return true
+}
+
+func (ccp *CosmosChainProcessor) logPacketMessage(message string, pi *packetInfo, additionalFields ...zap.Field) {
+	fields := []zap.Field{
+		zap.String("message", message),
+		zap.Uint64("sequence", pi.packet.Sequence),
+		zap.String("src_channel", pi.packet.SourceChannel),
+		zap.String("src_port", pi.packet.SourcePort),
+		zap.String("dst_channel", pi.packet.DestinationChannel),
+		zap.String("dst_port", pi.packet.DestinationPort),
+	}
+	fields = append(fields, additionalFields...)
+	ccp.log.Debug("Observed IBC message", fields...)
+}
+
+// END packet msg handlers
diff --git a/relayer/chains/cosmos/types.go b/relayer/chains/cosmos/types.go
index 886dda0..cc6126e 100644
--- a/relayer/chains/cosmos/types.go
+++ b/relayer/chains/cosmos/types.go
@@ -3,6 +3,7 @@ package cosmos
 import (
 	clienttypes "github.com/cosmos/ibc-go/v3/modules/core/02-client/types"
 	chantypes "github.com/cosmos/ibc-go/v3/modules/core/04-channel/types"
+	"github.com/cosmos/relayer/v2/relayer/processor"
 )
 
 // ibcMessage is the type used for parsing all possible properties of IBC messages
@@ -20,6 +21,16 @@ type channelInfo struct {
 	connectionID          string
 }
 
+// channelKey returns the processor.ChannelKey from channelInfo
+func (c channelInfo) channelKey() processor.ChannelKey {
+	return processor.ChannelKey{
+		ChannelID:             c.channelID,
+		PortID:                c.portID,
+		CounterpartyChannelID: c.counterpartyChannelID,
+		CounterpartyPortID:    c.counterpartyPortID,
+	}
+}
+
 type connectionInfo struct {
 	connectionID             string
 	clientID                 string
@@ -39,9 +50,29 @@ type packetInfo struct {
 	connectionID    string
 }
 
+// channelKey returns the processor.ChannelKey from packetInfo
+func (p packetInfo) channelKey() processor.ChannelKey {
+	return processor.ChannelKey{
+		ChannelID:             p.packet.SourceChannel,
+		PortID:                p.packet.SourcePort,
+		CounterpartyChannelID: p.packet.DestinationChannel,
+		CounterpartyPortID:    p.packet.DestinationPort,
+	}
+}
+
 // clientInfo contains the consensus height of the counterparty chain for a client.
 type clientInfo struct {
 	clientID        string
 	consensusHeight clienttypes.Height
 	header          []byte
 }
+
+type channelOpenState map[processor.ChannelKey]bool
+
+func (c channelOpenState) Clone() channelOpenState {
+	newChannelOpenState := make(channelOpenState, len(c))
+	for k, v := range c {
+		newChannelOpenState[k] = v
+	}
+	return newChannelOpenState
+}
diff --git a/relayer/chains/mock/message_handlers.go b/relayer/chains/mock/message_handlers.go
index cc5b46f..e5c405d 100644
--- a/relayer/chains/mock/message_handlers.go
+++ b/relayer/chains/mock/message_handlers.go
@@ -19,7 +19,7 @@ type MsgHandlerParams struct {
 var messageHandlers = map[string]func(MsgHandlerParams){
 	processor.MsgTransfer:        handleMsgTransfer,
 	processor.MsgRecvPacket:      handleMsgRecvPacket,
-	processor.MsgAcknowledgement: handleMsgAcknowlegement,
+	processor.MsgAcknowledgement: handleMsgAcknowledgement,
 
 	// TODO handlers for packet timeout, client, channel, and connection messages
 }
@@ -70,7 +70,7 @@ func handleMsgRecvPacket(p MsgHandlerParams) {
 	)
 }
 
-func handleMsgAcknowlegement(p MsgHandlerParams) {
+func handleMsgAcknowledgement(p MsgHandlerParams) {
 	channelKey := processor.ChannelKey{
 		ChannelID:             p.PacketInfo.SourceChannel,
 		PortID:                p.PacketInfo.SourcePort,
diff --git a/relayer/chains/mock/mock_chain_processor.go b/relayer/chains/mock/mock_chain_processor.go
index f8df20a..99e5db8 100644
--- a/relayer/chains/mock/mock_chain_processor.go
+++ b/relayer/chains/mock/mock_chain_processor.go
@@ -6,6 +6,7 @@ import (
 
 	"github.com/cosmos/relayer/v2/relayer/processor"
 	"github.com/cosmos/relayer/v2/relayer/provider"
+	"github.com/cosmos/relayer/v2/relayer/provider/cosmos"
 
 	chantypes "github.com/cosmos/ibc-go/v3/modules/core/04-channel/types"
 	"go.uber.org/zap"
@@ -50,7 +51,7 @@ func (mcp *MockChainProcessor) SetPathProcessors(pathProcessors processor.PathPr
 
 // Provider returns the ChainProvider, which provides the methods for querying, assembling IBC messages, and sending transactions.
 func (mcp *MockChainProcessor) Provider() provider.ChainProvider {
-	return nil
+	return &cosmos.CosmosProvider{PCfg: cosmos.CosmosProviderConfig{ChainID: mcp.chainID}}
 }
 
 type queryCyclePersistence struct {
@@ -148,17 +149,24 @@ func (mcp *MockChainProcessor) queryCycle(ctx context.Context, persistence *quer
 		}
 		// }
 
-		// now pass foundMessages to the path processors
-		for channelKey, messages := range foundMessages {
-			// TODO do not relay on closed channels
-			for _, pp := range mcp.pathProcessors {
-				mcp.log.Info("sending messages to path processor", zap.String("chain_id", mcp.chainID))
-				pp.HandleNewMessages(mcp.chainID, channelKey, processor.ChainProcessorCacheData{
-					NewMessages: messages,
-					InSync:      mcp.inSync,
-				})
+		channelStateCache := make(processor.ChannelStateCache)
+
+		// mocking all channels open
+		for channelKey := range foundMessages {
+			channelStateCache[channelKey] = processor.ChannelState{
+				Open: true,
 			}
 		}
+
+		// now pass foundMessages to the path processors
+		for _, pp := range mcp.pathProcessors {
+			mcp.log.Info("sending messages to path processor", zap.String("chain_id", mcp.chainID))
+			pp.HandleNewData(mcp.chainID, processor.ChainProcessorCacheData{
+				ChannelMessageCache: foundMessages,
+				InSync:              mcp.inSync,
+				ChannelStateCache:   channelStateCache,
+			})
+		}
 		persistence.latestQueriedBlock = i
 	}
 }
diff --git a/relayer/processor/path_end.go b/relayer/processor/path_end.go
index e64fa7b..fcb16cb 100644
--- a/relayer/processor/path_end.go
+++ b/relayer/processor/path_end.go
@@ -1,9 +1,14 @@
 package processor
 
 type PathEnd struct {
-	ChainID string
+	ChainID      string
+	ClientID     string
+	ConnectionID string
 
-	// TODO clientID, connectionID, and channel filter stuff (allowlist, denylist)
+	// TODO channel filter stuff (allowlist, denylist)
 }
 
-// TODO methods for determining if a given channel, client, or connection is relevant for this pathEnd
+func (pe PathEnd) ShouldRelayChannel(channelKey ChannelKey) bool {
+	// TODO return based on channel filter
+	return true
+}
diff --git a/relayer/processor/path_processor.go b/relayer/processor/path_processor.go
index e0f58de..f0c0da1 100644
--- a/relayer/processor/path_processor.go
+++ b/relayer/processor/path_processor.go
@@ -12,7 +12,8 @@ import (
 )
 
 const (
-	// DurationErrorRetry determines how long
+	// DurationErrorRetry determines how long to wait before retrying
+	// in the case of failure to send transactions with IBC messages.
 	DurationErrorRetry = 5 * time.Second
 )
 
@@ -31,6 +32,15 @@ type PathProcessor struct {
 // PathProcessors is a slice of PathProcessor instances
 type PathProcessors []*PathProcessor
 
+func (p PathProcessors) IsRelayedChannel(k ChannelKey, chainID string) bool {
+	for _, pp := range p {
+		if pp.IsRelayedChannel(chainID, k) {
+			return true
+		}
+	}
+	return false
+}
+
 // PathEndRuntime is used at runtime for each chain involved in the path.
 // It holds a channel for incoming messages from the ChainProcessors, which will
 // be processed during Run(ctx).
@@ -39,7 +49,9 @@ type PathEndRuntime struct {
 
 	chainProvider provider.ChainProvider
 
-	messageCache MessageCache
+	messageCache ChannelMessageCache
+
+	channelStateCache ChannelStateCache
 
 	// New messages and other data arriving from the handleNewMessagesForPathEnd method.
 	incomingCacheData chan ChainProcessorCacheData
@@ -48,6 +60,13 @@ type PathEndRuntime struct {
 	inSync bool
 }
 
+func (pathEnd *PathEndRuntime) MergeCacheData(d ChainProcessorCacheData) {
+	// TODO make sure passes channel filter for pathEnd1 before calling this
+	pathEnd.messageCache.Merge(d.ChannelMessageCache)    // Merge incoming packet IBC messages into the backlog
+	pathEnd.channelStateCache.Merge(d.ChannelStateCache) // Merge incoming channel state IBC messages into the backlog
+	pathEnd.inSync = d.InSync
+}
+
 // IBCMessageWithSequence holds a packet's sequence along with it,
 // useful for sending packets around internal to the PathProcessor.
 type IBCMessageWithSequence struct {
@@ -61,25 +80,59 @@ func NewPathProcessor(log *zap.Logger, pathEnd1 PathEnd, pathEnd2 PathEnd) *Path
 		pathEnd1: &PathEndRuntime{
 			info:              pathEnd1,
 			incomingCacheData: make(chan ChainProcessorCacheData, 100),
-			messageCache:      make(MessageCache),
+			channelStateCache: make(ChannelStateCache),
+			messageCache:      make(ChannelMessageCache),
 		},
 		pathEnd2: &PathEndRuntime{
 			info:              pathEnd2,
 			incomingCacheData: make(chan ChainProcessorCacheData, 100),
-			messageCache:      make(MessageCache),
+			channelStateCache: make(ChannelStateCache),
+			messageCache:      make(ChannelMessageCache),
 		},
 		retryProcess: make(chan struct{}, 8),
 	}
 }
 
 // TEST USE ONLY
-func (pp *PathProcessor) PathEnd1Messages(message string) SequenceCache {
-	return pp.pathEnd1.messageCache[message]
+func (pp *PathProcessor) PathEnd1Messages(channelKey ChannelKey, message string) SequenceCache {
+	return pp.pathEnd1.messageCache[channelKey][message]
 }
 
 // TEST USE ONLY
-func (pp *PathProcessor) PathEnd2Messages(message string) SequenceCache {
-	return pp.pathEnd2.messageCache[message]
+func (pp *PathProcessor) PathEnd2Messages(channelKey ChannelKey, message string) SequenceCache {
+	return pp.pathEnd2.messageCache[channelKey][message]
+}
+
+type channelPair struct {
+	pathEnd1ChannelKey ChannelKey
+	pathEnd2ChannelKey ChannelKey
+}
+
+func (pp *PathProcessor) channelPairs() []channelPair {
+	// Channel keys are from pathEnd1's perspective
+	channels := make(map[ChannelKey]bool)
+	for key, state := range pp.pathEnd1.channelStateCache {
+		if !state.Open {
+			continue
+		}
+		channels[key] = true
+	}
+	for key, state := range pp.pathEnd2.channelStateCache {
+		if !state.Open {
+			continue
+		}
+		channels[key.Counterparty()] = true
+	}
+	channelPairs := make([]channelPair, len(channels))
+	i := 0
+	for key := range channels {
+		channelPairs[i] = channelPair{
+			pathEnd1ChannelKey: key,
+			pathEnd2ChannelKey: key.Counterparty(),
+		}
+		i++
+	}
+	return channelPairs
 }
 
 // Path Processors are constructed before ChainProcessors, so reference needs to be added afterwards
@@ -98,6 +151,33 @@ func (pp *PathProcessor) SetChainProviderIfApplicable(chainProvider provider.Cha
 	return false
 }
 
+func (pp *PathProcessor) IsRelayedChannel(chainID string, channelKey ChannelKey) bool {
+	if pp.pathEnd1.info.ChainID == chainID {
+		return pp.pathEnd1.info.ShouldRelayChannel(channelKey)
+	} else if pp.pathEnd2.info.ChainID == chainID {
+		return pp.pathEnd2.info.ShouldRelayChannel(channelKey)
+	}
+	return false
+}
+
+func (pp *PathProcessor) IsRelevantClient(chainID string, clientID string) bool {
+	if pp.pathEnd1.info.ChainID == chainID && pp.pathEnd1.info.ClientID == clientID {
+		return true
+	} else if pp.pathEnd2.info.ChainID == chainID && pp.pathEnd2.info.ClientID == clientID {
+		return true
+	}
+	return false
+}
+
+func (pp *PathProcessor) IsRelevantConnection(chainID string, connectionID string) bool {
+	if pp.pathEnd1.info.ChainID == chainID && pp.pathEnd1.info.ConnectionID == connectionID {
+		return true
+	} else if pp.pathEnd2.info.ChainID == chainID && pp.pathEnd2.info.ConnectionID == connectionID {
+		return true
+	}
+	return false
+}
+
 // ProcessBacklogIfReady gives ChainProcessors a way to trigger the path processor process
 // as soon as they are in sync for the first time, even if they do not have new messages.
 func (pp *PathProcessor) ProcessBacklogIfReady() {
@@ -112,12 +192,10 @@ func (pp *PathProcessor) ProcessBacklogIfReady() {
 }
 
 // ChainProcessors call this method when they have new IBC messages
-func (pp *PathProcessor) HandleNewMessages(chainID string, channelKey ChannelKey, cacheData ChainProcessorCacheData) {
+func (pp *PathProcessor) HandleNewData(chainID string, cacheData ChainProcessorCacheData) {
 	if pp.pathEnd1.info.ChainID == chainID {
-		// TODO make sure passes channel filter for pathEnd1 before calling this
 		pp.pathEnd1.incomingCacheData <- cacheData
 	} else if pp.pathEnd2.info.ChainID == chainID {
-		// TODO make sure passes channel filter for pathEnd2 before calling this
 		pp.pathEnd2.incomingCacheData <- cacheData
 	}
 }
@@ -211,31 +289,47 @@ func (pp *PathProcessor) sendMessages(pathEnd *PathEndRuntime, messages []IBCMes
 
 // messages from both pathEnds are needed in order to determine what needs to be relayed for a single pathEnd
 func (pp *PathProcessor) processLatestMessages() error {
-	pathEnd1PacketFlowMessages := PathEndPacketFlowMessages{
-		SrcMsgTransfer:        pp.pathEnd1.messageCache[MsgTransfer],
-		DstMsgRecvPacket:      pp.pathEnd2.messageCache[MsgRecvPacket],
-		SrcMsgAcknowledgement: pp.pathEnd1.messageCache[MsgAcknowledgement],
-	}
-	pathEnd2PacketFlowMessages := PathEndPacketFlowMessages{
-		SrcMsgTransfer:        pp.pathEnd2.messageCache[MsgTransfer],
-		DstMsgRecvPacket:      pp.pathEnd1.messageCache[MsgRecvPacket],
-		SrcMsgAcknowledgement: pp.pathEnd2.messageCache[MsgAcknowledgement],
-	}
+	channelPairs := pp.channelPairs()
 
 	// process the packet flows for both packends to determine what needs to be relayed
-	var pathEnd1ProcessRes, pathEnd2ProcessRes PathEndProcessedResponse
+	pathEnd1ProcessRes := make([]*PathEndProcessedResponse, len(channelPairs))
+	pathEnd2ProcessRes := make([]*PathEndProcessedResponse, len(channelPairs))
+
 	var wg sync.WaitGroup
-	wg.Add(2)
-	go pp.getUnrelayedPacketsAndAcksAndToDelete(pathEnd1PacketFlowMessages, &wg, &pathEnd1ProcessRes)
-	go pp.getUnrelayedPacketsAndAcksAndToDelete(pathEnd2PacketFlowMessages, &wg, &pathEnd2ProcessRes)
+
+	for i, pair := range channelPairs {
+		pathEnd1PacketFlowMessages := PathEndPacketFlowMessages{
+			SrcMsgTransfer:        pp.pathEnd1.messageCache[pair.pathEnd1ChannelKey][MsgTransfer],
+			DstMsgRecvPacket:      pp.pathEnd2.messageCache[pair.pathEnd2ChannelKey][MsgRecvPacket],
+			SrcMsgAcknowledgement: pp.pathEnd1.messageCache[pair.pathEnd1ChannelKey][MsgAcknowledgement],
+		}
+		pathEnd2PacketFlowMessages := PathEndPacketFlowMessages{
+			SrcMsgTransfer:        pp.pathEnd2.messageCache[pair.pathEnd2ChannelKey][MsgTransfer],
+			DstMsgRecvPacket:      pp.pathEnd1.messageCache[pair.pathEnd1ChannelKey][MsgRecvPacket],
+			SrcMsgAcknowledgement: pp.pathEnd2.messageCache[pair.pathEnd2ChannelKey][MsgAcknowledgement],
+		}
+
+		pathEnd1ProcessRes[i] = new(PathEndProcessedResponse)
+		pathEnd2ProcessRes[i] = new(PathEndProcessedResponse)
+
+		wg.Add(2)
+		go pp.getUnrelayedPacketsAndAcksAndToDelete(pathEnd1PacketFlowMessages, &wg, pathEnd1ProcessRes[i])
+		go pp.getUnrelayedPacketsAndAcksAndToDelete(pathEnd2PacketFlowMessages, &wg, pathEnd2ProcessRes[i])
+	}
 	wg.Wait()
 
 	// concatenate applicable messages for pathend
-	pathEnd1Messages := append(pathEnd2ProcessRes.UnrelayedPackets, pathEnd1ProcessRes.UnrelayedAcknowledgements...)
-	pathEnd2Messages := append(pathEnd1ProcessRes.UnrelayedPackets, pathEnd2ProcessRes.UnrelayedAcknowledgements...)
+	var pathEnd1Messages, pathEnd2Messages []IBCMessageWithSequence
+	for i := 0; i < len(channelPairs); i++ {
+		pathEnd1Messages = append(pathEnd1Messages, pathEnd2ProcessRes[i].UnrelayedPackets...)
+		pathEnd1Messages = append(pathEnd1Messages, pathEnd1ProcessRes[i].UnrelayedAcknowledgements...)
 
-	pp.pathEnd1.messageCache.DeleteCachedMessages(pathEnd1ProcessRes.ToDeleteSrc, pathEnd2ProcessRes.ToDeleteDst)
-	pp.pathEnd2.messageCache.DeleteCachedMessages(pathEnd2ProcessRes.ToDeleteSrc, pathEnd1ProcessRes.ToDeleteDst)
+		pathEnd2Messages = append(pathEnd2Messages, pathEnd1ProcessRes[i].UnrelayedPackets...)
+		pathEnd2Messages = append(pathEnd2Messages, pathEnd2ProcessRes[i].UnrelayedAcknowledgements...)
+
+		pp.pathEnd1.messageCache[channelPairs[i].pathEnd1ChannelKey].DeleteCachedMessages(pathEnd1ProcessRes[i].ToDeleteSrc, pathEnd2ProcessRes[i].ToDeleteDst)
+		pp.pathEnd2.messageCache[channelPairs[i].pathEnd2ChannelKey].DeleteCachedMessages(pathEnd2ProcessRes[i].ToDeleteSrc, pathEnd1ProcessRes[i].ToDeleteDst)
+	}
 
 	// now send messages in parallel
 	var eg errgroup.Group
@@ -252,22 +346,21 @@ func (pp *PathProcessor) Run(ctx context.Context) {
 			return
 
 		case d := <-pp.pathEnd1.incomingCacheData:
-			// if new messages are available from pathEnd1, run processLatestMessages
-			pp.pathEnd1.messageCache.Merge(d.NewMessages) // Merge incoming messages into the backlog of IBC messages for pathEnd1
-			pp.pathEnd1.inSync = d.InSync
+			// we have new data from ChainProcessor for pathEnd1
+			pp.pathEnd1.MergeCacheData(d)
 
 		case d := <-pp.pathEnd2.incomingCacheData:
-			// if new messages are available from pathEnd2, run processLatestMessages
-			pp.pathEnd2.messageCache.Merge(d.NewMessages) // Merge incoming messages into the backlog of IBC messages for pathEnd2
-			pp.pathEnd2.inSync = d.InSync
+			// we have new data from ChainProcessor for pathEnd2
+			pp.pathEnd2.MergeCacheData(d)
 
 		case <-pp.retryProcess:
-			// No new messages to merge in, just retry handling.
+			// No new data to merge in, just retry handling.
 		}
 
 		if !pp.pathEnd1.inSync || !pp.pathEnd2.inSync {
 			continue
 		}
+
 		// process latest message cache state from both pathEnds
 		if err := pp.processLatestMessages(); err != nil {
 			// in case of IBC message send errors, schedule retry after DurationErrorRetry
diff --git a/relayer/processor/types.go b/relayer/processor/types.go
index ca9064e..df0ad52 100644
--- a/relayer/processor/types.go
+++ b/relayer/processor/types.go
@@ -58,9 +58,62 @@ type ChannelKey struct {
 	CounterpartyPortID    string
 }
 
+// Counterparty flips a ChannelKey for the perspective of the counterparty chain
+func (channelKey ChannelKey) Counterparty() ChannelKey {
+	return ChannelKey{
+		ChannelID:             channelKey.CounterpartyChannelID,
+		PortID:                channelKey.CounterpartyPortID,
+		CounterpartyChannelID: channelKey.ChannelID,
+		CounterpartyPortID:    channelKey.PortID,
+	}
+}
+
+// ChannelState is used for tracking the most recent state of a channel.
+type ChannelState struct {
+	// Is the channel currently open, i.e. can packets be relayed on this channel now.
+	Open bool
+	// Any new IBC messages relevant to the channel so that the PathProcessor
+	// can take necessary action such as complete the channel handshake.
+	Messages []string
+}
+
+// ChannelStateCache maintains channel state for multiple channels.
+type ChannelStateCache map[ChannelKey]ChannelState
+
+// Flush is used to make a copy of the ChannelStateCache, helpful for the ChainProcessors
+// to clone their local cache so that PathProcessors can have a thread-safe copy.
+// It will also empty the messages for the existing channel state.
+func (c ChannelStateCache) Flush() ChannelStateCache {
+	copy := make(ChannelStateCache)
+	for channelKey, existingState := range c {
+		copy[channelKey] = existingState
+		flushed := existingState
+		flushed.Messages = nil
+		c[channelKey] = flushed
+	}
+	return copy
+}
+
+// Merge will merge another ChannelStateCache into this one, appending messages and updating the Open state.
+func (c ChannelStateCache) Merge(other ChannelStateCache) {
+	for channelKey, newState := range other {
+		existingState, ok := c[channelKey]
+		if !ok {
+			c[channelKey] = newState
+			return
+		}
+		existingState.Messages = append(existingState.Messages, newState.Messages...)
+		existingState.Open = newState.Open
+		c[channelKey] = existingState
+	}
+}
+
+// ChainProcessorCacheData is the data sent from the ChainProcessors to the PathProcessors
+// to keep the PathProcessors up to date with the latest info from the chains.
 type ChainProcessorCacheData struct {
-	NewMessages MessageCache
-	InSync      bool
+	ChannelMessageCache
+	InSync bool
+	ChannelStateCache
 }
 
 // Merge will merge another MessageCache into this one.
@@ -92,3 +145,50 @@ func (c MessageCache) DeleteCachedMessages(toDelete ...map[string][]uint64) {
 		}
 	}
 }
+
+// Merge will merge another MessageCache into this one.
+func (c ChannelMessageCache) Merge(other ChannelMessageCache) {
+	for channelKey, messageCache := range other {
+		_, ok := c[channelKey]
+		if !ok {
+			c[channelKey] = messageCache
+		} else {
+			c[channelKey].Merge(messageCache)
+		}
+	}
+}
+
+// ShouldRetainSequence returns true if packet is applicable to the channels for path processors that are subscribed to this chain processor
+func (c ChannelMessageCache) ShouldRetainSequence(p PathProcessors, k ChannelKey, chainID string, m string, seq uint64) bool {
+	if !p.IsRelayedChannel(k, chainID) {
+		return false
+	}
+	if _, ok := c[k]; !ok {
+		return true
+	}
+	if _, ok := c[k][m]; !ok {
+		return true
+	}
+	for sequence := range c[k][m] {
+		if sequence == seq {
+			// already have this sequence number
+			// there can be multiple MsgRecvPacket, MsgAcknowledgement, MsgTimeout, and MsgTimeoutOnClose for the same packet
+			// from different relayers.
+			return false
+		}
+	}
+
+	return true
+}
+
+// Retain assumes the packet is applicable to the channels for a path processor that is subscribed to this chain processor.
+// It creates cache path if it doesn't exist, then caches message.
+func (c ChannelMessageCache) Retain(k ChannelKey, m string, seq uint64, ibcMsg provider.RelayerMessage) {
+	if _, ok := c[k]; !ok {
+		c[k] = make(MessageCache)
+	}
+	if _, ok := c[k][m]; !ok {
+		c[k][m] = make(SequenceCache)
+	}
+	c[k][m][seq] = ibcMsg
+}


Test patch:
diff --git a/relayer/chains/cosmos/message_handlers_test.go b/relayer/chains/cosmos/message_handlers_test.go
new file mode 100644
index 0000000..2ece934
--- /dev/null
+++ b/relayer/chains/cosmos/message_handlers_test.go
@@ -0,0 +1,299 @@
+package cosmos
+
+import (
+	"os"
+	"testing"
+
+	clienttypes "github.com/cosmos/ibc-go/v3/modules/core/02-client/types"
+	chantypes "github.com/cosmos/ibc-go/v3/modules/core/04-channel/types"
+	"github.com/cosmos/relayer/v2/relayer/processor"
+	"github.com/cosmos/relayer/v2/relayer/provider/cosmos"
+	"github.com/google/go-cmp/cmp"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+)
+
+func mockCosmosChainProcessor(t *testing.T) *CosmosChainProcessor {
+	const (
+		chainID1 = "test-chain-1"
+		chainID2 = "test-chain-2"
+	)
+	var (
+		pathEnd1      = processor.PathEnd{ChainID: chainID1}
+		pathEnd2      = processor.PathEnd{ChainID: chainID2}
+		log           = zap.NewNop()
+		pathProcessor = processor.NewPathProcessor(log, pathEnd1, pathEnd2)
+		provider      = cosmos.CosmosProvider{PCfg: cosmos.CosmosProviderConfig{ChainID: chainID1}}
+	)
+
+	ccp, err := NewCosmosChainProcessor(log, &provider, "", os.Stdin, os.Stdout, []*processor.PathProcessor{pathProcessor})
+	require.NoError(t, err, "error constructing cosmos chain processor")
+
+	applicable := pathProcessor.SetChainProviderIfApplicable(&provider)
+	require.True(t, applicable, "error setting path processor reference to chain processor")
+
+	return ccp
+}
+
+func TestHandleMsgTransfer(t *testing.T) {
+	const (
+		sequence         = uint64(1)
+		srcChannel       = "channel-0"
+		dstChannel       = "channel-1"
+		srcPort          = "transfer"
+		dstPort          = "transfer"
+		timeoutHeight    = 100
+		timeoutRevision  = 1
+		timeoutTimestamp = uint64(2054566111724000000)
+	)
+	var (
+		packetData = []byte{0x1, 0x2, 0x3, 0x4}
+		ccp        = mockCosmosChainProcessor(t)
+	)
+
+	foundMessages := make(processor.ChannelMessageCache)
+
+	packetInfo := &packetInfo{
+		packet: chantypes.Packet{
+			Data:               packetData,
+			Sequence:           sequence,
+			SourceChannel:      srcChannel,
+			SourcePort:         srcPort,
+			DestinationChannel: dstChannel,
+			DestinationPort:    dstPort,
+			TimeoutHeight: clienttypes.Height{
+				RevisionHeight: timeoutHeight,
+				RevisionNumber: timeoutRevision,
+			},
+			TimeoutTimestamp: timeoutTimestamp,
+		},
+	}
+
+	ccp.handleMsgTransfer(MsgHandlerParams{messageInfo: packetInfo, foundMessages: foundMessages})
+
+	require.Len(t, foundMessages, 1)
+
+	channelKey := packetInfo.channelKey()
+
+	channelMessages, ok := foundMessages[channelKey]
+	require.True(t, ok, "unable to find messages for channel key")
+
+	require.Len(t, channelMessages, 1)
+
+	transferMessages, ok := channelMessages[processor.MsgTransfer]
+	require.True(t, ok, "unable to find messages for MsgTransfer type")
+
+	require.Len(t, transferMessages, 1)
+
+	sequenceMessage, ok := transferMessages[sequence]
+	require.True(t, ok, "unable to find message for sequence")
+
+	cosmosMsg := cosmos.CosmosMsg(sequenceMessage)
+	require.NotNil(t, cosmosMsg, "error parsing message as CosmosMsg")
+
+	msgRecvPacket, ok := cosmosMsg.(*chantypes.MsgRecvPacket)
+	require.True(t, ok, "unable to read message as MsgRecvPacket")
+
+	require.Empty(t, cmp.Diff(packetInfo.packet, msgRecvPacket.Packet), "MsgRecvPacket data does not match MsgTransfer data")
+}
+
+func TestHandleMsgRecvPacket(t *testing.T) {
+	const (
+		sequence   = uint64(1)
+		srcChannel = "channel-0"
+		dstChannel = "channel-1"
+		srcPort    = "transfer"
+		dstPort    = "transfer"
+	)
+	var (
+		packetData = []byte{0x1, 0x2, 0x3, 0x4}
+		packetAck  = []byte{0x2, 0x3, 0x4, 0x5}
+		ccp        = mockCosmosChainProcessor(t)
+	)
+
+	foundMessages := make(processor.ChannelMessageCache)
+
+	packetInfo := &packetInfo{
+		packet: chantypes.Packet{
+			Data:               packetData,
+			Sequence:           sequence,
+			SourceChannel:      srcChannel,
+			SourcePort:         srcPort,
+			DestinationChannel: dstChannel,
+			DestinationPort:    dstPort,
+		},
+		ack: packetAck,
+	}
+
+	ccp.handleMsgRecvPacket(MsgHandlerParams{messageInfo: packetInfo, foundMessages: foundMessages})
+
+	require.Len(t, foundMessages, 1)
+
+	// flipped on purpose since MsgRecvPacket is committed on counterparty chain
+	channelKey := packetInfo.channelKey().Counterparty()
+
+	channelMessages, ok := foundMessages[channelKey]
+	require.True(t, ok, "unable to find messages for channel key")
+
+	require.Len(t, channelMessages, 1)
+
+	transferMessages, ok := channelMessages[processor.MsgRecvPacket]
+	require.True(t, ok, "unable to find messages for MsgRecvPacket type")
+
+	require.Len(t, transferMessages, 1)
+
+	sequenceMessage, ok := transferMessages[sequence]
+	require.True(t, ok, "unable to find message for sequence")
+
+	cosmosMsg := cosmos.CosmosMsg(sequenceMessage)
+	require.NotNil(t, cosmosMsg, "error parsing message as CosmosMsg")
+
+	msgRecvPacket, ok := cosmosMsg.(*chantypes.MsgAcknowledgement)
+	require.True(t, ok, "unable to read message as MsgAcknowledgement")
+
+	require.Empty(t, cmp.Diff(packetInfo.packet, msgRecvPacket.Packet), "MsgAcknowledgement data does not match MsgRecvPacket data")
+}
+
+func TestHandleMsgAcknowledgement(t *testing.T) {
+	const (
+		sequence   = uint64(1)
+		srcChannel = "channel-0"
+		dstChannel = "channel-1"
+		srcPort    = "transfer"
+		dstPort    = "transfer"
+	)
+	var (
+		packetData = []byte{0x1, 0x2, 0x3, 0x4}
+		ccp        = mockCosmosChainProcessor(t)
+	)
+
+	foundMessages := make(processor.ChannelMessageCache)
+
+	packetInfo := &packetInfo{
+		packet: chantypes.Packet{
+			Data:               packetData,
+			Sequence:           sequence,
+			SourceChannel:      srcChannel,
+			SourcePort:         srcPort,
+			DestinationChannel: dstChannel,
+			DestinationPort:    dstPort,
+		},
+	}
+
+	ccp.handleMsgAcknowledgement(MsgHandlerParams{messageInfo: packetInfo, foundMessages: foundMessages})
+
+	require.Len(t, foundMessages, 1)
+
+	channelKey := packetInfo.channelKey()
+
+	channelMessages, ok := foundMessages[channelKey]
+	require.True(t, ok, "unable to find messages for channel key")
+
+	require.Len(t, channelMessages, 1)
+
+	transferMessages, ok := channelMessages[processor.MsgAcknowledgement]
+	require.True(t, ok, "unable to find messages for MsgAcknowledgement type")
+
+	require.Len(t, transferMessages, 1)
+
+	sequenceMessage, ok := transferMessages[sequence]
+	require.True(t, ok, "unable to find message for sequence")
+
+	require.Nil(t, sequenceMessage, "message is not nil, expected nil since no messages need to be constructed for counterparty")
+}
+
+func TestHandleMsgTimeout(t *testing.T) {
+	const (
+		sequence   = uint64(1)
+		srcChannel = "channel-0"
+		dstChannel = "channel-1"
+		srcPort    = "transfer"
+		dstPort    = "transfer"
+	)
+	var (
+		packetData = []byte{0x1, 0x2, 0x3, 0x4}
+		ccp        = mockCosmosChainProcessor(t)
+	)
+
+	foundMessages := make(processor.ChannelMessageCache)
+
+	packetInfo := &packetInfo{
+		packet: chantypes.Packet{
+			Data:               packetData,
+			Sequence:           sequence,
+			SourceChannel:      srcChannel,
+			SourcePort:         srcPort,
+			DestinationChannel: dstChannel,
+			DestinationPort:    dstPort,
+		},
+	}
+
+	ccp.handleMsgTimeout(MsgHandlerParams{messageInfo: packetInfo, foundMessages: foundMessages})
+
+	require.Len(t, foundMessages, 1)
+
+	channelKey := packetInfo.channelKey()
+
+	channelMessages, ok := foundMessages[channelKey]
+	require.True(t, ok, "unable to find messages for channel key")
+
+	require.Len(t, channelMessages, 1)
+
+	transferMessages, ok := channelMessages[processor.MsgTimeout]
+	require.True(t, ok, "unable to find messages for MsgTimeout type")
+
+	require.Len(t, transferMessages, 1)
+
+	sequenceMessage, ok := transferMessages[sequence]
+	require.True(t, ok, "unable to find message for sequence")
+
+	require.Nil(t, sequenceMessage, "message is not nil, expected nil since no messages need to be constructed for counterparty")
+}
+
+func TestHandleMsgTimeoutOnClose(t *testing.T) {
+	const (
+		sequence   = uint64(1)
+		srcChannel = "channel-0"
+		dstChannel = "channel-1"
+		srcPort    = "transfer"
+		dstPort    = "transfer"
+	)
+	var (
+		packetData = []byte{0x1, 0x2, 0x3, 0x4}
+		ccp        = mockCosmosChainProcessor(t)
+	)
+
+	foundMessages := make(processor.ChannelMessageCache)
+
+	packetInfo := &packetInfo{
+		packet: chantypes.Packet{
+			Data:               packetData,
+			Sequence:           sequence,
+			SourceChannel:      srcChannel,
+			SourcePort:         srcPort,
+			DestinationChannel: dstChannel,
+			DestinationPort:    dstPort,
+		},
+	}
+
+	ccp.handleMsgTimeoutOnClose(MsgHandlerParams{messageInfo: packetInfo, foundMessages: foundMessages})
+
+	require.Len(t, foundMessages, 1)
+
+	channelKey := packetInfo.channelKey()
+
+	channelMessages, ok := foundMessages[channelKey]
+	require.True(t, ok, "unable to find messages for channel key")
+
+	require.Len(t, channelMessages, 1)
+
+	transferMessages, ok := channelMessages[processor.MsgTimeoutOnClose]
+	require.True(t, ok, "unable to find messages for MsgTimeoutOnClose type")
+
+	require.Len(t, transferMessages, 1)
+
+	sequenceMessage, ok := transferMessages[sequence]
+	require.True(t, ok, "unable to find message for sequence")
+
+	require.Nil(t, sequenceMessage, "message is not nil, expected nil since no messages need to be constructed for counterparty")
+}
diff --git a/relayer/chains/mock/mock_chain_processor_test.go b/relayer/chains/mock/mock_chain_processor_test.go
index 72a6469..d9ebdeb 100644
--- a/relayer/chains/mock/mock_chain_processor_test.go
+++ b/relayer/chains/mock/mock_chain_processor_test.go
@@ -31,11 +31,20 @@ func TestMockChainAndPathProcessors(t *testing.T) {
 	lastSentMockMsgRecvSequence2 := uint64(0)
 	var mockLock sync.Mutex
 
+	mockChannelKey1 := processor.ChannelKey{
+		ChannelID:             "channel-0",
+		PortID:                "port-0",
+		CounterpartyChannelID: "channel-1",
+		CounterpartyPortID:    "port-1",
+	}
+
+	mockChannelKey2 := mockChannelKey1.Counterparty()
+
 	getMockMessages1 := func() []mock.TransactionMessage {
-		return getMockMessages(&mockSequence1, &mockSequence2, &lastSentMockMsgRecvSequence2, &mockLock)
+		return getMockMessages(mockChannelKey1, &mockSequence1, &mockSequence2, &lastSentMockMsgRecvSequence2, &mockLock)
 	}
 	getMockMessages2 := func() []mock.TransactionMessage {
-		return getMockMessages(&mockSequence2, &mockSequence1, &lastSentMockMsgRecvSequence1, &mockLock)
+		return getMockMessages(mockChannelKey2, &mockSequence2, &mockSequence1, &lastSentMockMsgRecvSequence1, &mockLock)
 	}
 
 	log := zaptest.NewLogger(t)
@@ -57,13 +66,13 @@ func TestMockChainAndPathProcessors(t *testing.T) {
 	err := eventProcessor.Run(ctx)
 	require.NoError(t, err, "error running event processor")
 
-	pathEnd1LeftoverMsgTransfer := pathProcessor.PathEnd1Messages(processor.MsgTransfer)
-	pathEnd1LeftoverMsgRecvPacket := pathProcessor.PathEnd1Messages(processor.MsgRecvPacket)
-	pathEnd1LeftoverMsgAcknowledgement := pathProcessor.PathEnd1Messages(processor.MsgAcknowledgement)
+	pathEnd1LeftoverMsgTransfer := pathProcessor.PathEnd1Messages(mockChannelKey1, processor.MsgTransfer)
+	pathEnd1LeftoverMsgRecvPacket := pathProcessor.PathEnd1Messages(mockChannelKey1, processor.MsgRecvPacket)
+	pathEnd1LeftoverMsgAcknowledgement := pathProcessor.PathEnd1Messages(mockChannelKey1, processor.MsgAcknowledgement)
 
-	pathEnd2LeftoverMsgTransfer := pathProcessor.PathEnd2Messages(processor.MsgTransfer)
-	pathEnd2LeftoverMsgRecvPacket := pathProcessor.PathEnd2Messages(processor.MsgRecvPacket)
-	pathEnd2LeftoverMsgAcknowledgement := pathProcessor.PathEnd2Messages(processor.MsgAcknowledgement)
+	pathEnd2LeftoverMsgTransfer := pathProcessor.PathEnd2Messages(mockChannelKey2, processor.MsgTransfer)
+	pathEnd2LeftoverMsgRecvPacket := pathProcessor.PathEnd2Messages(mockChannelKey2, processor.MsgRecvPacket)
+	pathEnd2LeftoverMsgAcknowledgement := pathProcessor.PathEnd2Messages(mockChannelKey2, processor.MsgAcknowledgement)
 
 	log.Debug("leftover",
 		zap.Int("pathEnd1MsgTransfer", len(pathEnd1LeftoverMsgTransfer)),
@@ -90,27 +99,45 @@ func TestMockChainAndPathProcessors(t *testing.T) {
 // MsgTransfer
 // MsgRecvPacket for counterparty
 // MsgAcknowledgement
-func getMockMessages(mockSequence, mockSequenceCounterparty, lastSentMockMsgRecvCounterparty *uint64, lock *sync.Mutex) []mock.TransactionMessage {
+func getMockMessages(channelKey processor.ChannelKey, mockSequence, mockSequenceCounterparty, lastSentMockMsgRecvCounterparty *uint64, lock *sync.Mutex) []mock.TransactionMessage {
 	lock.Lock()
 	defer lock.Unlock()
 	*mockSequence++
 	mockMessages := []mock.TransactionMessage{
 		{
-			Action:     processor.MsgTransfer,
-			PacketInfo: &chantypes.Packet{Sequence: *mockSequence},
+			Action: processor.MsgTransfer,
+			PacketInfo: &chantypes.Packet{
+				Sequence:           *mockSequence,
+				SourceChannel:      channelKey.ChannelID,
+				SourcePort:         channelKey.PortID,
+				DestinationChannel: channelKey.CounterpartyChannelID,
+				DestinationPort:    channelKey.CounterpartyPortID,
+			},
 		},
 	}
 	if *mockSequenceCounterparty > 1 && *lastSentMockMsgRecvCounterparty != *mockSequenceCounterparty {
 		*lastSentMockMsgRecvCounterparty = *mockSequenceCounterparty
 		mockMessages = append(mockMessages, mock.TransactionMessage{
-			Action:     processor.MsgRecvPacket,
-			PacketInfo: &chantypes.Packet{Sequence: *mockSequenceCounterparty - 1},
+			Action: processor.MsgRecvPacket,
+			PacketInfo: &chantypes.Packet{
+				Sequence:           *mockSequenceCounterparty - 1,
+				SourceChannel:      channelKey.CounterpartyChannelID,
+				SourcePort:         channelKey.CounterpartyPortID,
+				DestinationChannel: channelKey.ChannelID,
+				DestinationPort:    channelKey.PortID,
+			},
 		})
 	}
 	if *mockSequence > 2 {
 		mockMessages = append(mockMessages, mock.TransactionMessage{
-			Action:     processor.MsgAcknowledgement,
-			PacketInfo: &chantypes.Packet{Sequence: *mockSequence - 2},
+			Action: processor.MsgAcknowledgement,
+			PacketInfo: &chantypes.Packet{
+				Sequence:           *mockSequence - 2,
+				SourceChannel:      channelKey.ChannelID,
+				SourcePort:         channelKey.PortID,
+				DestinationChannel: channelKey.CounterpartyChannelID,
+				DestinationPort:    channelKey.CounterpartyPortID,
+			},
 		})
 	}
 	return mockMessages