Repository: cosmos/relayer
Pull request #796 (2022-06-14 22:19:04)

CosmosChainProcessor Client message handlers
Adds `latestClientState` caching on the `CosmosChainProcessor`, which will be used in a future PR for constructing trusted IBC headers so that `MsgUpdateClient` messages can be constructed for the counterparty chain.

Changes cache of data shared between `ChainProcessor` and `PathProcessor` from `ChannelMessageCache` to `IBCMessagesCache` to account for not only packet flow messages, but also connection and channel handshake messages, which the `PathProcessor` can handle, if applicable, to complete the connection and channel handshakes (in a future PR).

Adds `Merge` and `Retain` helpers for the different cached data types.


Patch:
diff --git a/relayer/chains/cosmos/cosmos_chain_processor.go b/relayer/chains/cosmos/cosmos_chain_processor.go
index 89e0b02..34e51ec 100644
--- a/relayer/chains/cosmos/cosmos_chain_processor.go
+++ b/relayer/chains/cosmos/cosmos_chain_processor.go
@@ -28,7 +28,10 @@ type CosmosChainProcessor struct {
 	// indicates whether queries are in sync with latest height of the chain
 	inSync bool
 
-	// holds open state for discovered channels
+	// holds highest consensus height and header for all clients
+	latestClientState
+
+	// holds open state for known channels
 	channelStateCache processor.ChannelStateCache
 }
 
@@ -42,6 +45,7 @@ func NewCosmosChainProcessor(log *zap.Logger, provider *cosmos.CosmosProvider, r
 		chainProvider:     provider,
 		cc:                cc,
 		pathProcessors:    pathProcessors,
+		latestClientState: make(latestClientState),
 		channelStateCache: make(processor.ChannelStateCache),
 	}, nil
 }
@@ -55,6 +59,31 @@ const (
 	inSyncNumBlocksThreshold    = 2
 )
 
+type msgHandlerParams struct {
+	// incoming IBC message
+	messageInfo interface{}
+
+	// reference to the caches that will be assembled by the handlers in this file
+	ibcMessagesCache processor.IBCMessagesCache
+}
+
+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,
+
+	processor.MsgCreateClient:       (*CosmosChainProcessor).handleMsgCreateClient,
+	processor.MsgUpdateClient:       (*CosmosChainProcessor).handleMsgUpdateClient,
+	processor.MsgUpgradeClient:      (*CosmosChainProcessor).handleMsgUpgradeClient,
+	processor.MsgSubmitMisbehaviour: (*CosmosChainProcessor).handleMsgSubmitMisbehaviour,
+}
+
+func (ccp *CosmosChainProcessor) logObservedIBCMessage(m string, fields ...zap.Field) {
+	ccp.log.With(zap.String("message", m)).Debug("Observed IBC message", fields...)
+}
+
 // Provider returns the ChainProvider, which provides the methods for querying, assembling IBC messages, and sending transactions.
 func (ccp *CosmosChainProcessor) Provider() provider.ChainProvider {
 	return ccp.chainProvider
@@ -178,7 +207,7 @@ func (ccp *CosmosChainProcessor) queryCycle(ctx context.Context, persistence *qu
 		}
 	}
 
-	foundMessages := make(processor.ChannelMessageCache)
+	ibcMessagesCache := processor.NewIBCMessagesCache()
 
 	ppChanged := false
 
@@ -204,9 +233,9 @@ func (ccp *CosmosChainProcessor) queryCycle(ctx context.Context, persistence *qu
 					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,
+				changed := handler(ccp, msgHandlerParams{
+					messageInfo:      m.messageInfo,
+					ibcMessagesCache: ibcMessagesCache,
 				})
 				ppChanged = ppChanged || changed
 			}
@@ -227,10 +256,9 @@ func (ccp *CosmosChainProcessor) queryCycle(ctx context.Context, persistence *qu
 
 	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(),
+			IBCMessagesCache:  ibcMessagesCache,
+			InSync:            ccp.inSync,
+			ChannelStateCache: ccp.channelStateCache,
 		})
 	}
 
diff --git a/relayer/chains/cosmos/msg_handlers_client.go b/relayer/chains/cosmos/msg_handlers_client.go
new file mode 100644
index 0000000..f8d7218
--- /dev/null
+++ b/relayer/chains/cosmos/msg_handlers_client.go
@@ -0,0 +1,37 @@
+package cosmos
+
+import (
+	"go.uber.org/zap"
+)
+
+func (ccp *CosmosChainProcessor) handleMsgCreateClient(p msgHandlerParams) bool {
+	clientInfo := p.messageInfo.(*clientInfo)
+	// save the latest consensus height and header for this client
+	ccp.latestClientState.UpdateLatestClientState(*clientInfo)
+	ccp.logObservedIBCMessage("MsgCreateClient", zap.String("client_id", clientInfo.clientID))
+	return false
+}
+
+func (ccp *CosmosChainProcessor) handleMsgUpdateClient(p msgHandlerParams) bool {
+	clientInfo := p.messageInfo.(*clientInfo)
+	// save the latest consensus height and header for this client
+	ccp.latestClientState.UpdateLatestClientState(*clientInfo)
+	ccp.logObservedIBCMessage("MsgUpdateClient", zap.String("client_id", clientInfo.clientID))
+	return false
+}
+
+func (ccp *CosmosChainProcessor) handleMsgUpgradeClient(p msgHandlerParams) bool {
+	clientInfo := p.messageInfo.(*clientInfo)
+	// save the latest consensus height and header for this client
+	ccp.latestClientState.UpdateLatestClientState(*clientInfo)
+	ccp.logObservedIBCMessage("MsgUpgradeClient", zap.String("client_id", clientInfo.clientID))
+	return false
+}
+
+func (ccp *CosmosChainProcessor) handleMsgSubmitMisbehaviour(p msgHandlerParams) bool {
+	clientInfo := p.messageInfo.(*clientInfo)
+	// save the latest consensus height and header for this client
+	ccp.latestClientState.UpdateLatestClientState(*clientInfo)
+	ccp.logObservedIBCMessage("MsgSubmitMisbehaviour", zap.String("client_id", clientInfo.clientID))
+	return false
+}
diff --git a/relayer/chains/cosmos/message_handlers.go b/relayer/chains/cosmos/msg_handlers_packet.go
similarity index 67%
rename from relayer/chains/cosmos/message_handlers.go
rename to relayer/chains/cosmos/msg_handlers_packet.go
index 0a53f0a..74ddda9 100644
--- a/relayer/chains/cosmos/message_handlers.go
+++ b/relayer/chains/cosmos/msg_handlers_packet.go
@@ -7,28 +7,12 @@ import (
 	"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 {
+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) {
+	if !p.ibcMessagesCache.PacketFlow.ShouldRetainSequence(ccp.pathProcessors, channelKey, ccp.chainProvider.ChainId(), processor.MsgTransfer, pi.packet.Sequence) {
 		return false
 	}
 	// Construct the start of the MsgRecvPacket for the counterparty chain.
@@ -37,7 +21,7 @@ func (ccp *CosmosChainProcessor) handleMsgTransfer(p MsgHandlerParams) bool {
 	// 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{
+	p.ibcMessagesCache.PacketFlow.Retain(channelKey, processor.MsgTransfer, pi.packet.Sequence, cosmos.NewCosmosMessage(&chantypes.MsgRecvPacket{
 		Packet: chantypes.Packet{
 			Sequence:           pi.packet.Sequence,
 			SourcePort:         pi.packet.SourcePort,
@@ -57,12 +41,12 @@ func (ccp *CosmosChainProcessor) handleMsgTransfer(p MsgHandlerParams) bool {
 	return true
 }
 
-func (ccp *CosmosChainProcessor) handleMsgRecvPacket(p MsgHandlerParams) bool {
+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) {
+	if !p.ibcMessagesCache.PacketFlow.ShouldRetainSequence(ccp.pathProcessors, channelKey, ccp.chainProvider.ChainId(), processor.MsgRecvPacket, pi.packet.Sequence) {
 		return false
 	}
 	// Construct the start of the MsgAcknowledgement for the counterparty chain.
@@ -70,7 +54,7 @@ func (ccp *CosmosChainProcessor) handleMsgRecvPacket(p MsgHandlerParams) bool {
 	// 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{
+	p.ibcMessagesCache.PacketFlow.Retain(channelKey, processor.MsgRecvPacket, pi.packet.Sequence, cosmos.NewCosmosMessage(&chantypes.MsgAcknowledgement{
 		Packet: chantypes.Packet{
 			Sequence:           pi.packet.Sequence,
 			SourcePort:         pi.packet.SourcePort,
@@ -87,56 +71,55 @@ func (ccp *CosmosChainProcessor) handleMsgRecvPacket(p MsgHandlerParams) bool {
 	return true
 }
 
-func (ccp *CosmosChainProcessor) handleMsgAcknowledgement(p MsgHandlerParams) bool {
+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) {
+	if !p.ibcMessagesCache.PacketFlow.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)
+	p.ibcMessagesCache.PacketFlow.Retain(channelKey, processor.MsgAcknowledgement, pi.packet.Sequence, nil)
 	ccp.logPacketMessage("MsgAcknowledgement", pi)
 	return true
 }
 
-func (ccp *CosmosChainProcessor) handleMsgTimeout(p MsgHandlerParams) bool {
+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) {
+	if !p.ibcMessagesCache.PacketFlow.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)
+	p.ibcMessagesCache.PacketFlow.Retain(channelKey, processor.MsgTimeout, pi.packet.Sequence, nil)
 	ccp.logPacketMessage("MsgTimeout", pi)
 	return true
 }
 
-func (ccp *CosmosChainProcessor) handleMsgTimeoutOnClose(p MsgHandlerParams) bool {
+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) {
+	if !p.ibcMessagesCache.PacketFlow.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)
+	p.ibcMessagesCache.PacketFlow.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),
@@ -144,7 +127,5 @@ func (ccp *CosmosChainProcessor) logPacketMessage(message string, pi *packetInfo
 		zap.String("dst_port", pi.packet.DestinationPort),
 	}
 	fields = append(fields, additionalFields...)
-	ccp.log.Debug("Observed IBC message", fields...)
+	ccp.logObservedIBCMessage(message, fields...)
 }
-
-// END packet msg handlers
diff --git a/relayer/chains/cosmos/types.go b/relayer/chains/cosmos/types.go
index cc6126e..4af9b0c 100644
--- a/relayer/chains/cosmos/types.go
+++ b/relayer/chains/cosmos/types.go
@@ -38,6 +38,15 @@ type connectionInfo struct {
 	counterpartyConnectionID string
 }
 
+func (c connectionInfo) connectionKey() processor.ConnectionKey {
+	return processor.ConnectionKey{
+		ConnectionID:             c.connectionID,
+		ClientID:                 c.clientID,
+		CounterpartyConnectionID: c.counterpartyConnectionID,
+		CounterpartyClientID:     c.counterpartyClientID,
+	}
+}
+
 // packetInfo contains pertinent packet information for constructing IBC messages for the counterparty.
 type packetInfo struct {
 	// Packet is the IBC conformant Packet.
@@ -67,6 +76,29 @@ type clientInfo struct {
 	header          []byte
 }
 
+// latestClientState is a map of clientID to the latest clientInfo for that client.
+type latestClientState map[string]clientInfo
+
+func (l latestClientState) UpdateLatestClientState(clientInfo clientInfo) {
+	existingClientInfo, ok := l[clientInfo.clientID]
+	if ok && clientInfo.consensusHeight.LT(existingClientInfo.consensusHeight) {
+		// height is less than latest, so no-op
+		return
+	}
+
+	// update latest if no existing state or provided consensus height is newer
+	l[clientInfo.clientID] = clientInfo
+}
+
+func (l latestClientState) Clone() latestClientState {
+	newLatestClientState := make(latestClientState, len(l))
+	for k, v := range l {
+		newLatestClientState[k] = v
+	}
+	return newLatestClientState
+}
+
+// channelOpenState is a map of channelKey to the latest open state, true/false, for a channel.
 type channelOpenState map[processor.ChannelKey]bool
 
 func (c channelOpenState) Clone() channelOpenState {
diff --git a/relayer/chains/mock/message_handlers.go b/relayer/chains/mock/message_handlers.go
index e5c405d..5acf4f7 100644
--- a/relayer/chains/mock/message_handlers.go
+++ b/relayer/chains/mock/message_handlers.go
@@ -5,18 +5,17 @@ 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"
 	"github.com/cosmos/relayer/v2/relayer/provider/cosmos"
 	"go.uber.org/zap"
 )
 
-type MsgHandlerParams struct {
-	mcp           *MockChainProcessor
-	PacketInfo    *chantypes.Packet
-	FoundMessages processor.ChannelMessageCache
+type msgHandlerParams struct {
+	mcp              *MockChainProcessor
+	packetInfo       *chantypes.Packet
+	ibcMessagesCache processor.IBCMessagesCache
 }
 
-var messageHandlers = map[string]func(MsgHandlerParams){
+var messageHandlers = map[string]func(msgHandlerParams){
 	processor.MsgTransfer:        handleMsgTransfer,
 	processor.MsgRecvPacket:      handleMsgRecvPacket,
 	processor.MsgAcknowledgement: handleMsgAcknowledgement,
@@ -24,67 +23,56 @@ var messageHandlers = map[string]func(MsgHandlerParams){
 	// TODO handlers for packet timeout, client, channel, and connection messages
 }
 
-func retainMessage(p MsgHandlerParams, channelKey processor.ChannelKey, message string, sequence uint64, ibcMessage provider.RelayerMessage) {
-	if _, ok := p.FoundMessages[channelKey]; !ok {
-		p.FoundMessages[channelKey] = make(processor.MessageCache)
-	}
-	if _, ok := p.FoundMessages[channelKey][message]; !ok {
-		p.FoundMessages[channelKey][message] = make(processor.SequenceCache)
-	}
-	p.FoundMessages[channelKey][message][sequence] = ibcMessage
-}
-
-func handleMsgTransfer(p MsgHandlerParams) {
+func handleMsgTransfer(p msgHandlerParams) {
 	channelKey := processor.ChannelKey{
-		ChannelID:             p.PacketInfo.SourceChannel,
-		PortID:                p.PacketInfo.SourcePort,
-		CounterpartyChannelID: p.PacketInfo.DestinationChannel,
-		CounterpartyPortID:    p.PacketInfo.DestinationPort,
+		ChannelID:             p.packetInfo.SourceChannel,
+		PortID:                p.packetInfo.SourcePort,
+		CounterpartyChannelID: p.packetInfo.DestinationChannel,
+		CounterpartyPortID:    p.packetInfo.DestinationPort,
 	}
-	retainMessage(p, channelKey, processor.MsgTransfer, p.PacketInfo.Sequence, cosmos.NewCosmosMessage(&chantypes.MsgRecvPacket{Packet: *p.PacketInfo}))
+	p.ibcMessagesCache.PacketFlow.Retain(channelKey, processor.MsgTransfer, p.packetInfo.Sequence, cosmos.NewCosmosMessage(&chantypes.MsgRecvPacket{Packet: *p.packetInfo}))
 	p.mcp.log.Debug("observed MsgTransfer",
 		zap.String("chain_id", p.mcp.chainID),
-		zap.Uint64("sequence", p.PacketInfo.Sequence),
-		zap.String("src_channel", p.PacketInfo.SourceChannel),
-		zap.String("src_port", p.PacketInfo.SourcePort),
-		zap.String("timeout_height", fmt.Sprintf("%d-%d", p.PacketInfo.TimeoutHeight.RevisionNumber, p.PacketInfo.TimeoutHeight.RevisionHeight)),
-		zap.Uint64("timeout_timestamp", p.PacketInfo.TimeoutTimestamp),
+		zap.Uint64("sequence", p.packetInfo.Sequence),
+		zap.String("src_channel", p.packetInfo.SourceChannel),
+		zap.String("src_port", p.packetInfo.SourcePort),
+		zap.String("timeout_height", fmt.Sprintf("%d-%d", p.packetInfo.TimeoutHeight.RevisionNumber, p.packetInfo.TimeoutHeight.RevisionHeight)),
+		zap.Uint64("timeout_timestamp", p.packetInfo.TimeoutTimestamp),
 	)
 }
 
-func handleMsgRecvPacket(p MsgHandlerParams) {
+func handleMsgRecvPacket(p msgHandlerParams) {
 	channelKey := processor.ChannelKey{
-		ChannelID:             p.PacketInfo.DestinationChannel,
-		PortID:                p.PacketInfo.DestinationPort,
-		CounterpartyChannelID: p.PacketInfo.SourceChannel,
-		CounterpartyPortID:    p.PacketInfo.SourcePort,
+		ChannelID:             p.packetInfo.DestinationChannel,
+		PortID:                p.packetInfo.DestinationPort,
+		CounterpartyChannelID: p.packetInfo.SourceChannel,
+		CounterpartyPortID:    p.packetInfo.SourcePort,
 	}
-	retainMessage(p, channelKey, processor.MsgRecvPacket, p.PacketInfo.Sequence, cosmos.NewCosmosMessage(&chantypes.MsgAcknowledgement{Packet: *p.PacketInfo}))
+	p.ibcMessagesCache.PacketFlow.Retain(channelKey, processor.MsgRecvPacket, p.packetInfo.Sequence, cosmos.NewCosmosMessage(&chantypes.MsgAcknowledgement{Packet: *p.packetInfo}))
 	p.mcp.log.Debug("observed MsgRecvPacket",
 		zap.String("chain_id", p.mcp.chainID),
-		zap.Uint64("sequence", p.PacketInfo.Sequence),
-		zap.String("src_channel", p.PacketInfo.SourceChannel),
-		zap.String("src_port", p.PacketInfo.SourcePort),
-		zap.String("dst_channel", p.PacketInfo.DestinationChannel),
-		zap.String("dst_port", p.PacketInfo.DestinationPort),
+		zap.Uint64("sequence", p.packetInfo.Sequence),
+		zap.String("src_channel", p.packetInfo.SourceChannel),
+		zap.String("src_port", p.packetInfo.SourcePort),
+		zap.String("dst_channel", p.packetInfo.DestinationChannel),
+		zap.String("dst_port", p.packetInfo.DestinationPort),
 	)
 }
 
-func handleMsgAcknowledgement(p MsgHandlerParams) {
+func handleMsgAcknowledgement(p msgHandlerParams) {
 	channelKey := processor.ChannelKey{
-		ChannelID:             p.PacketInfo.SourceChannel,
-		PortID:                p.PacketInfo.SourcePort,
-		CounterpartyChannelID: p.PacketInfo.DestinationChannel,
-		CounterpartyPortID:    p.PacketInfo.DestinationPort,
+		ChannelID:             p.packetInfo.SourceChannel,
+		PortID:                p.packetInfo.SourcePort,
+		CounterpartyChannelID: p.packetInfo.DestinationChannel,
+		CounterpartyPortID:    p.packetInfo.DestinationPort,
 	}
-	retainMessage(p, channelKey, processor.MsgAcknowledgement, p.PacketInfo.Sequence, nil)
-
+	p.ibcMessagesCache.PacketFlow.Retain(channelKey, processor.MsgAcknowledgement, p.packetInfo.Sequence, nil)
 	p.mcp.log.Debug("observed MsgAcknowledgement",
 		zap.String("chain_id", p.mcp.chainID),
-		zap.Uint64("sequence", p.PacketInfo.Sequence),
-		zap.String("src_channel", p.PacketInfo.SourceChannel),
-		zap.String("src_port", p.PacketInfo.SourcePort),
-		zap.String("dst_channel", p.PacketInfo.DestinationChannel),
-		zap.String("dst_port", p.PacketInfo.DestinationPort),
+		zap.Uint64("sequence", p.packetInfo.Sequence),
+		zap.String("src_channel", p.packetInfo.SourceChannel),
+		zap.String("src_port", p.packetInfo.SourcePort),
+		zap.String("dst_channel", p.packetInfo.DestinationChannel),
+		zap.String("dst_port", p.packetInfo.DestinationPort),
 	)
 }
diff --git a/relayer/chains/mock/mock_chain_processor.go b/relayer/chains/mock/mock_chain_processor.go
index 99e5db8..08caeed 100644
--- a/relayer/chains/mock/mock_chain_processor.go
+++ b/relayer/chains/mock/mock_chain_processor.go
@@ -122,7 +122,7 @@ func (mcp *MockChainProcessor) queryCycle(ctx context.Context, persistence *quer
 		// fetch block
 
 		// used for collecting IBC messages that will be sent to the Path Processors
-		foundMessages := make(processor.ChannelMessageCache)
+		ibcMessagesCache := processor.NewIBCMessagesCache()
 
 		// iterate through transactions
 		// iterate through messages in transactions
@@ -140,10 +140,10 @@ func (mcp *MockChainProcessor) queryCycle(ctx context.Context, persistence *quer
 		// this can be parralelized also
 		for _, m := range messages {
 			if handler, ok := messageHandlers[m.Action]; ok {
-				handler(MsgHandlerParams{
-					mcp:           mcp,
-					PacketInfo:    m.PacketInfo,
-					FoundMessages: foundMessages,
+				handler(msgHandlerParams{
+					mcp:              mcp,
+					packetInfo:       m.PacketInfo,
+					ibcMessagesCache: ibcMessagesCache,
 				})
 			}
 		}
@@ -152,19 +152,17 @@ func (mcp *MockChainProcessor) queryCycle(ctx context.Context, persistence *quer
 		channelStateCache := make(processor.ChannelStateCache)
 
 		// mocking all channels open
-		for channelKey := range foundMessages {
-			channelStateCache[channelKey] = processor.ChannelState{
-				Open: true,
-			}
+		for channelKey := range ibcMessagesCache.PacketFlow {
+			channelStateCache[channelKey] = 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,
+				IBCMessagesCache:  ibcMessagesCache,
+				InSync:            mcp.inSync,
+				ChannelStateCache: channelStateCache,
 			})
 		}
 		persistence.latestQueriedBlock = i
diff --git a/relayer/processor/path_processor.go b/relayer/processor/path_processor.go
index f0c0da1..cc7b646 100644
--- a/relayer/processor/path_processor.go
+++ b/relayer/processor/path_processor.go
@@ -49,8 +49,8 @@ type PathEndRuntime struct {
 
 	chainProvider provider.ChainProvider
 
-	messageCache ChannelMessageCache
-
+	// cached data
+	messageCache      IBCMessagesCache
 	channelStateCache ChannelStateCache
 
 	// New messages and other data arriving from the handleNewMessagesForPathEnd method.
@@ -62,8 +62,8 @@ type PathEndRuntime struct {
 
 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.messageCache.Merge(d.IBCMessagesCache)       // Merge incoming packet IBC messages into the backlog
+	pathEnd.channelStateCache.Merge(d.ChannelStateCache) // Update latest channel open state for chain
 	pathEnd.inSync = d.InSync
 }
 
@@ -81,26 +81,26 @@ func NewPathProcessor(log *zap.Logger, pathEnd1 PathEnd, pathEnd2 PathEnd) *Path
 			info:              pathEnd1,
 			incomingCacheData: make(chan ChainProcessorCacheData, 100),
 			channelStateCache: make(ChannelStateCache),
-			messageCache:      make(ChannelMessageCache),
+			messageCache:      NewIBCMessagesCache(),
 		},
 		pathEnd2: &PathEndRuntime{
 			info:              pathEnd2,
 			incomingCacheData: make(chan ChainProcessorCacheData, 100),
 			channelStateCache: make(ChannelStateCache),
-			messageCache:      make(ChannelMessageCache),
+			messageCache:      NewIBCMessagesCache(),
 		},
 		retryProcess: make(chan struct{}, 8),
 	}
 }
 
 // TEST USE ONLY
-func (pp *PathProcessor) PathEnd1Messages(channelKey ChannelKey, message string) SequenceCache {
-	return pp.pathEnd1.messageCache[channelKey][message]
+func (pp *PathProcessor) PathEnd1Messages(channelKey ChannelKey, message string) PacketSequenceCache {
+	return pp.pathEnd1.messageCache.PacketFlow[channelKey][message]
 }
 
 // TEST USE ONLY
-func (pp *PathProcessor) PathEnd2Messages(channelKey ChannelKey, message string) SequenceCache {
-	return pp.pathEnd2.messageCache[channelKey][message]
+func (pp *PathProcessor) PathEnd2Messages(channelKey ChannelKey, message string) PacketSequenceCache {
+	return pp.pathEnd2.messageCache.PacketFlow[channelKey][message]
 }
 
 type channelPair struct {
@@ -111,28 +111,25 @@ type channelPair struct {
 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 k, open := range pp.pathEnd1.channelStateCache {
+		channels[k] = open
 	}
-	for key, state := range pp.pathEnd2.channelStateCache {
-		if !state.Open {
-			continue
-		}
-		channels[key.Counterparty()] = true
+	for k, open := range pp.pathEnd2.channelStateCache {
+		channels[k.Counterparty()] = open
 	}
-	channelPairs := make([]channelPair, len(channels))
+	pairs := make([]channelPair, len(channels))
 	i := 0
-	for key := range channels {
-		channelPairs[i] = channelPair{
-			pathEnd1ChannelKey: key,
-			pathEnd2ChannelKey: key.Counterparty(),
+	for k, open := range channels {
+		if !open {
+			continue
+		}
+		pairs[i] = channelPair{
+			pathEnd1ChannelKey: k,
+			pathEnd2ChannelKey: k.Counterparty(),
 		}
 		i++
 	}
-	return channelPairs
+	return pairs
 }
 
 // Path Processors are constructed before ChainProcessors, so reference needs to be added afterwards
@@ -203,17 +200,17 @@ func (pp *PathProcessor) HandleNewData(chainID string, cacheData ChainProcessorC
 // this contains MsgRecvPacket from same chain
 // needs to be transformed into PathEndPacketFlowMessages once counterparty info is available to complete packet flow state for pathEnd
 type PathEndMessages struct {
-	MsgTransfer        SequenceCache
-	MsgRecvPacket      SequenceCache
-	MsgAcknowledgement SequenceCache
+	MsgTransfer        PacketSequenceCache
+	MsgRecvPacket      PacketSequenceCache
+	MsgAcknowledgement PacketSequenceCache
 }
 
 // contains MsgRecvPacket from counterparty
 // entire packet flow
 type PathEndPacketFlowMessages struct {
-	SrcMsgTransfer        SequenceCache
-	DstMsgRecvPacket      SequenceCache
-	SrcMsgAcknowledgement SequenceCache
+	SrcMsgTransfer        PacketSequenceCache
+	DstMsgRecvPacket      PacketSequenceCache
+	SrcMsgAcknowledgement PacketSequenceCache
 	// TODO SrcTimeout and SrcTimeoutOnClose
 }
 
@@ -299,14 +296,14 @@ func (pp *PathProcessor) processLatestMessages() error {
 
 	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],
+			SrcMsgTransfer:        pp.pathEnd1.messageCache.PacketFlow[pair.pathEnd1ChannelKey][MsgTransfer],
+			DstMsgRecvPacket:      pp.pathEnd2.messageCache.PacketFlow[pair.pathEnd2ChannelKey][MsgRecvPacket],
+			SrcMsgAcknowledgement: pp.pathEnd1.messageCache.PacketFlow[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],
+			SrcMsgTransfer:        pp.pathEnd2.messageCache.PacketFlow[pair.pathEnd2ChannelKey][MsgTransfer],
+			DstMsgRecvPacket:      pp.pathEnd1.messageCache.PacketFlow[pair.pathEnd1ChannelKey][MsgRecvPacket],
+			SrcMsgAcknowledgement: pp.pathEnd2.messageCache.PacketFlow[pair.pathEnd2ChannelKey][MsgAcknowledgement],
 		}
 
 		pathEnd1ProcessRes[i] = new(PathEndProcessedResponse)
@@ -327,8 +324,8 @@ func (pp *PathProcessor) processLatestMessages() error {
 		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)
+		pp.pathEnd1.messageCache.PacketFlow[channelPairs[i].pathEnd1ChannelKey].DeleteCachedMessages(pathEnd1ProcessRes[i].ToDeleteSrc, pathEnd2ProcessRes[i].ToDeleteDst)
+		pp.pathEnd2.messageCache.PacketFlow[channelPairs[i].pathEnd2ChannelKey].DeleteCachedMessages(pathEnd2ProcessRes[i].ToDeleteSrc, pathEnd1ProcessRes[i].ToDeleteDst)
 	}
 
 	// now send messages in parallel
diff --git a/relayer/processor/types.go b/relayer/processor/types.go
index df0ad52..bda6792 100644
--- a/relayer/processor/types.go
+++ b/relayer/processor/types.go
@@ -41,16 +41,42 @@ var (
 	MsgSubmitMisbehaviour = "/" + proto.MessageName((*clienttypes.MsgSubmitMisbehaviour)(nil))
 )
 
-// SequenceCache is used for caching an IBC message for a given packet sequence.
-type SequenceCache map[uint64]provider.RelayerMessage
+type IBCMessagesCache struct {
+	PacketFlow          ChannelPacketMessagesCache
+	ConnectionHandshake ConnectionMessagesCache
+	ChannelHandshake    ChannelMessagesCache
+}
+
+func NewIBCMessagesCache() IBCMessagesCache {
+	return IBCMessagesCache{
+		PacketFlow:          make(ChannelPacketMessagesCache),
+		ConnectionHandshake: make(ConnectionMessagesCache),
+		ChannelHandshake:    make(ChannelMessagesCache),
+	}
+}
+
+// ChannelPacketMessagesCache is used for caching a PacketMessagesCache for a given IBC channel.
+type ChannelPacketMessagesCache map[ChannelKey]PacketMessagesCache
+
+// PacketMessagesCache is used for caching a PacketSequenceCache for a given IBC message type.
+type PacketMessagesCache map[string]PacketSequenceCache
+
+// PacketSequenceCache is used for caching an IBC message for a given packet sequence.
+type PacketSequenceCache map[uint64]provider.RelayerMessage
+
+// ChannelMessagesCache is used for caching a ChannelMessageCache for a given IBC channel.
+type ChannelMessagesCache map[ChannelKey]ChannelMessageCache
+
+// ChannelMessageCache is used for caching channel handshake IBC messages for a given IBC message type.
+type ChannelMessageCache map[string]provider.RelayerMessage
 
-// MessageCache is used for caching a SequenceCache for a given IBC message type.
-type MessageCache map[string]SequenceCache
+// ConnectionMessagesCache is used for caching a ConnectionMessageCache for a given IBC connection.
+type ConnectionMessagesCache map[ConnectionKey]ConnectionMessageCache
 
-// ChannelMessageCache is used for caching a MessageCache for a given IBC channel.
-type ChannelMessageCache map[ChannelKey]MessageCache
+// ConnectionMessageCache is used for caching connection handshake IBC messages for a given IBC message type.
+type ConnectionMessageCache map[string]provider.RelayerMessage
 
-// ChannelKey is the key used between ChainProcessor and PathProcessor.
+// ChannelKey is the key used for identifying channels between ChainProcessor and PathProcessor.
 type ChannelKey struct {
 	ChannelID             string
 	PortID                string
@@ -68,75 +94,67 @@ func (channelKey ChannelKey) Counterparty() ChannelKey {
 	}
 }
 
-// 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
+// ConnectionKey is the key used for identifying connections between ChainProcessor and PathProcessor.
+type ConnectionKey struct {
+	ClientID                 string
+	ConnectionID             string
+	CounterpartyClientID     string
+	CounterpartyConnectionID 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
+// Counterparty flips a ConnectionKey for the perspective of the counterparty chain
+func (connectionKey ConnectionKey) Counterparty() ConnectionKey {
+	return ConnectionKey{
+		ClientID:                 connectionKey.CounterpartyClientID,
+		ConnectionID:             connectionKey.CounterpartyConnectionID,
+		CounterpartyClientID:     connectionKey.ClientID,
+		CounterpartyConnectionID: connectionKey.ConnectionID,
 	}
-	return copy
 }
 
+// ChannelStateCache maintains channel open state for multiple channels.
+type ChannelStateCache map[ChannelKey]bool
+
 // 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
+		c[channelKey] = newState
+	}
+}
+
+// ConnectionStateCache maintains connection open state for multiple connections.
+type ConnectionStateCache map[ConnectionKey]bool
+
+// Merge will merge another ChannelStateCache into this one, appending messages and updating the Open state.
+func (c ConnectionStateCache) Merge(other ConnectionStateCache) {
+	for channelKey, newState := range other {
+		c[channelKey] = newState
 	}
 }
 
 // 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 {
-	ChannelMessageCache
+	IBCMessagesCache
 	InSync bool
+	ConnectionStateCache
 	ChannelStateCache
 }
 
-// Merge will merge another MessageCache into this one.
-func (c MessageCache) Merge(other MessageCache) {
-	for k, v := range other {
-		c[k] = v
-	}
-}
-
-// Clone will create a deep copy of a MessageCache.
-func (c MessageCache) Clone() MessageCache {
-	newMessageCache := make(MessageCache)
+// Clone will create a deep copy of a PacketMessagesCache.
+func (c PacketMessagesCache) Clone() PacketMessagesCache {
+	newPacketMessagesCache := make(PacketMessagesCache, len(c))
 	for mk, mv := range c {
-		newSequenceCache := make(SequenceCache)
+		newPacketSequenceCache := make(PacketSequenceCache, len(mv))
 		for sk, sv := range mv {
-			newSequenceCache[sk] = sv
+			newPacketSequenceCache[sk] = sv
 		}
-		newMessageCache[mk] = newSequenceCache
+		newPacketMessagesCache[mk] = newPacketSequenceCache
 	}
-	return newMessageCache
+	return newPacketMessagesCache
 }
 
-func (c MessageCache) DeleteCachedMessages(toDelete ...map[string][]uint64) {
+func (c PacketMessagesCache) DeleteCachedMessages(toDelete ...map[string][]uint64) {
 	for _, toDeleteMap := range toDelete {
 		for message, toDeleteMessages := range toDeleteMap {
 			for _, sequence := range toDeleteMessages {
@@ -146,8 +164,14 @@ func (c MessageCache) DeleteCachedMessages(toDelete ...map[string][]uint64) {
 	}
 }
 
-// Merge will merge another MessageCache into this one.
-func (c ChannelMessageCache) Merge(other ChannelMessageCache) {
+func (c IBCMessagesCache) Merge(other IBCMessagesCache) {
+	c.ConnectionHandshake.Merge(other.ConnectionHandshake)
+	c.ChannelHandshake.Merge(other.ChannelHandshake)
+	c.PacketFlow.Merge(other.PacketFlow)
+}
+
+// Merge will merge another ChannelPacketMessagesCache into this one.
+func (c ChannelPacketMessagesCache) Merge(other ChannelPacketMessagesCache) {
 	for channelKey, messageCache := range other {
 		_, ok := c[channelKey]
 		if !ok {
@@ -159,7 +183,7 @@ func (c ChannelMessageCache) Merge(other ChannelMessageCache) {
 }
 
 // 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 {
+func (c ChannelPacketMessagesCache) ShouldRetainSequence(p PathProcessors, k ChannelKey, chainID string, m string, seq uint64) bool {
 	if !p.IsRelayedChannel(k, chainID) {
 		return false
 	}
@@ -183,12 +207,85 @@ func (c ChannelMessageCache) ShouldRetainSequence(p PathProcessors, k ChannelKey
 
 // 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) {
+func (c ChannelPacketMessagesCache) Retain(k ChannelKey, m string, seq uint64, ibcMsg provider.RelayerMessage) {
 	if _, ok := c[k]; !ok {
-		c[k] = make(MessageCache)
+		c[k] = make(PacketMessagesCache)
 	}
 	if _, ok := c[k][m]; !ok {
-		c[k][m] = make(SequenceCache)
+		c[k][m] = make(PacketSequenceCache)
 	}
 	c[k][m][seq] = ibcMsg
 }
+
+// Merge will merge another PacketMessagesCache into this one.
+func (c PacketMessagesCache) Merge(other PacketMessagesCache) {
+	for ibcMessage, messageCache := range other {
+		_, ok := c[ibcMessage]
+		if !ok {
+			c[ibcMessage] = messageCache
+		} else {
+			c[ibcMessage].Merge(messageCache)
+		}
+	}
+}
+
+// Merge will merge another PacketSequenceCache into this one.
+func (c PacketSequenceCache) Merge(other PacketSequenceCache) {
+	for k, v := range other {
+		c[k] = v
+	}
+}
+
+// Merge will merge another ConnectionMessagesCache into this one.
+func (c ConnectionMessagesCache) Merge(other ConnectionMessagesCache) {
+	for ibcMessage, messageCache := range other {
+		_, ok := c[ibcMessage]
+		if !ok {
+			c[ibcMessage] = messageCache
+		} else {
+			c[ibcMessage].Merge(messageCache)
+		}
+	}
+}
+
+// Retain assumes creates cache path if it doesn't exist, then caches message.
+func (c ConnectionMessagesCache) Retain(k ConnectionKey, m string, ibcMsg provider.RelayerMessage) {
+	if _, ok := c[k]; !ok {
+		c[k] = make(ConnectionMessageCache)
+	}
+	c[k][m] = ibcMsg
+}
+
+// Merge will merge another ConnectionMessageCache into this one.
+func (c ConnectionMessageCache) Merge(other ConnectionMessageCache) {
+	for k, v := range other {
+		c[k] = v
+	}
+}
+
+// Merge will merge another ChannelMessagesCache into this one.
+func (c ChannelMessagesCache) Merge(other ChannelMessagesCache) {
+	for ibcMessage, messageCache := range other {
+		_, ok := c[ibcMessage]
+		if !ok {
+			c[ibcMessage] = messageCache
+		} else {
+			c[ibcMessage].Merge(messageCache)
+		}
+	}
+}
+
+// Retain assumes creates cache path if it doesn't exist, then caches message.
+func (c ChannelMessagesCache) Retain(k ChannelKey, m string, ibcMsg provider.RelayerMessage) {
+	if _, ok := c[k]; !ok {
+		c[k] = make(ChannelMessageCache)
+	}
+	c[k][m] = ibcMsg
+}
+
+// Merge will merge another ChannelMessageCache into this one.
+func (c ChannelMessageCache) Merge(other ChannelMessageCache) {
+	for k, v := range other {
+		c[k] = v
+	}
+}


Test patch:
diff --git a/relayer/chains/cosmos/message_handlers_test.go b/relayer/chains/cosmos/msg_handlers_packet_test.go
similarity index 85%
rename from relayer/chains/cosmos/message_handlers_test.go
rename to relayer/chains/cosmos/msg_handlers_packet_test.go
index 2ece934..5ca9a4c 100644
--- a/relayer/chains/cosmos/message_handlers_test.go
+++ b/relayer/chains/cosmos/msg_handlers_packet_test.go
@@ -51,7 +51,7 @@ func TestHandleMsgTransfer(t *testing.T) {
 		ccp        = mockCosmosChainProcessor(t)
 	)
 
-	foundMessages := make(processor.ChannelMessageCache)
+	ibcMessagesCache := processor.NewIBCMessagesCache()
 
 	packetInfo := &packetInfo{
 		packet: chantypes.Packet{
@@ -69,13 +69,13 @@ func TestHandleMsgTransfer(t *testing.T) {
 		},
 	}
 
-	ccp.handleMsgTransfer(MsgHandlerParams{messageInfo: packetInfo, foundMessages: foundMessages})
+	ccp.handleMsgTransfer(msgHandlerParams{messageInfo: packetInfo, ibcMessagesCache: ibcMessagesCache})
 
-	require.Len(t, foundMessages, 1)
+	require.Len(t, ibcMessagesCache.PacketFlow, 1)
 
 	channelKey := packetInfo.channelKey()
 
-	channelMessages, ok := foundMessages[channelKey]
+	channelMessages, ok := ibcMessagesCache.PacketFlow[channelKey]
 	require.True(t, ok, "unable to find messages for channel key")
 
 	require.Len(t, channelMessages, 1)
@@ -111,7 +111,7 @@ func TestHandleMsgRecvPacket(t *testing.T) {
 		ccp        = mockCosmosChainProcessor(t)
 	)
 
-	foundMessages := make(processor.ChannelMessageCache)
+	ibcMessagesCache := processor.NewIBCMessagesCache()
 
 	packetInfo := &packetInfo{
 		packet: chantypes.Packet{
@@ -125,14 +125,14 @@ func TestHandleMsgRecvPacket(t *testing.T) {
 		ack: packetAck,
 	}
 
-	ccp.handleMsgRecvPacket(MsgHandlerParams{messageInfo: packetInfo, foundMessages: foundMessages})
+	ccp.handleMsgRecvPacket(msgHandlerParams{messageInfo: packetInfo, ibcMessagesCache: ibcMessagesCache})
 
-	require.Len(t, foundMessages, 1)
+	require.Len(t, ibcMessagesCache.PacketFlow, 1)
 
 	// flipped on purpose since MsgRecvPacket is committed on counterparty chain
 	channelKey := packetInfo.channelKey().Counterparty()
 
-	channelMessages, ok := foundMessages[channelKey]
+	channelMessages, ok := ibcMessagesCache.PacketFlow[channelKey]
 	require.True(t, ok, "unable to find messages for channel key")
 
 	require.Len(t, channelMessages, 1)
@@ -167,7 +167,7 @@ func TestHandleMsgAcknowledgement(t *testing.T) {
 		ccp        = mockCosmosChainProcessor(t)
 	)
 
-	foundMessages := make(processor.ChannelMessageCache)
+	ibcMessagesCache := processor.NewIBCMessagesCache()
 
 	packetInfo := &packetInfo{
 		packet: chantypes.Packet{
@@ -180,13 +180,13 @@ func TestHandleMsgAcknowledgement(t *testing.T) {
 		},
 	}
 
-	ccp.handleMsgAcknowledgement(MsgHandlerParams{messageInfo: packetInfo, foundMessages: foundMessages})
+	ccp.handleMsgAcknowledgement(msgHandlerParams{messageInfo: packetInfo, ibcMessagesCache: ibcMessagesCache})
 
-	require.Len(t, foundMessages, 1)
+	require.Len(t, ibcMessagesCache.PacketFlow, 1)
 
 	channelKey := packetInfo.channelKey()
 
-	channelMessages, ok := foundMessages[channelKey]
+	channelMessages, ok := ibcMessagesCache.PacketFlow[channelKey]
 	require.True(t, ok, "unable to find messages for channel key")
 
 	require.Len(t, channelMessages, 1)
@@ -215,7 +215,7 @@ func TestHandleMsgTimeout(t *testing.T) {
 		ccp        = mockCosmosChainProcessor(t)
 	)
 
-	foundMessages := make(processor.ChannelMessageCache)
+	ibcMessagesCache := processor.NewIBCMessagesCache()
 
 	packetInfo := &packetInfo{
 		packet: chantypes.Packet{
@@ -228,13 +228,13 @@ func TestHandleMsgTimeout(t *testing.T) {
 		},
 	}
 
-	ccp.handleMsgTimeout(MsgHandlerParams{messageInfo: packetInfo, foundMessages: foundMessages})
+	ccp.handleMsgTimeout(msgHandlerParams{messageInfo: packetInfo, ibcMessagesCache: ibcMessagesCache})
 
-	require.Len(t, foundMessages, 1)
+	require.Len(t, ibcMessagesCache.PacketFlow, 1)
 
 	channelKey := packetInfo.channelKey()
 
-	channelMessages, ok := foundMessages[channelKey]
+	channelMessages, ok := ibcMessagesCache.PacketFlow[channelKey]
 	require.True(t, ok, "unable to find messages for channel key")
 
 	require.Len(t, channelMessages, 1)
@@ -263,7 +263,7 @@ func TestHandleMsgTimeoutOnClose(t *testing.T) {
 		ccp        = mockCosmosChainProcessor(t)
 	)
 
-	foundMessages := make(processor.ChannelMessageCache)
+	ibcMessagesCache := processor.NewIBCMessagesCache()
 
 	packetInfo := &packetInfo{
 		packet: chantypes.Packet{
@@ -276,13 +276,13 @@ func TestHandleMsgTimeoutOnClose(t *testing.T) {
 		},
 	}
 
-	ccp.handleMsgTimeoutOnClose(MsgHandlerParams{messageInfo: packetInfo, foundMessages: foundMessages})
+	ccp.handleMsgTimeoutOnClose(msgHandlerParams{messageInfo: packetInfo, ibcMessagesCache: ibcMessagesCache})
 
-	require.Len(t, foundMessages, 1)
+	require.Len(t, ibcMessagesCache.PacketFlow, 1)
 
 	channelKey := packetInfo.channelKey()
 
-	channelMessages, ok := foundMessages[channelKey]
+	channelMessages, ok := ibcMessagesCache.PacketFlow[channelKey]
 	require.True(t, ok, "unable to find messages for channel key")
 
 	require.Len(t, channelMessages, 1)