Repository: cosmos/relayer
Pull request #799 (2022-06-15 22:11:48)

CosmosChainProcessor Channel message handlers
Channel handshake message handlers.

Similar to connection handshakes, maintains `channelOpenState` so that `PathProcessor` can avoid relaying packets on closed channels.


Patch:
diff --git a/relayer/chains/cosmos/cosmos_chain_processor.go b/relayer/chains/cosmos/cosmos_chain_processor.go
index 2ea4668..613b8ea 100644
--- a/relayer/chains/cosmos/cosmos_chain_processor.go
+++ b/relayer/chains/cosmos/cosmos_chain_processor.go
@@ -87,6 +87,13 @@ var messageHandlers = map[string]func(*CosmosChainProcessor, msgHandlerParams) b
 	processor.MsgConnectionOpenTry:     (*CosmosChainProcessor).handleMsgConnectionOpenTry,
 	processor.MsgConnectionOpenAck:     (*CosmosChainProcessor).handleMsgConnectionOpenAck,
 	processor.MsgConnectionOpenConfirm: (*CosmosChainProcessor).handleMsgConnectionOpenConfirm,
+
+	processor.MsgChannelCloseConfirm: (*CosmosChainProcessor).handleMsgChannelCloseConfirm,
+	processor.MsgChannelCloseInit:    (*CosmosChainProcessor).handleMsgChannelCloseInit,
+	processor.MsgChannelOpenAck:      (*CosmosChainProcessor).handleMsgChannelOpenAck,
+	processor.MsgChannelOpenConfirm:  (*CosmosChainProcessor).handleMsgChannelOpenConfirm,
+	processor.MsgChannelOpenInit:     (*CosmosChainProcessor).handleMsgChannelOpenInit,
+	processor.MsgChannelOpenTry:      (*CosmosChainProcessor).handleMsgChannelOpenTry,
 }
 
 func (ccp *CosmosChainProcessor) logObservedIBCMessage(m string, fields ...zap.Field) {
diff --git a/relayer/chains/cosmos/msg_handlers_channel.go b/relayer/chains/cosmos/msg_handlers_channel.go
new file mode 100644
index 0000000..6f55c8a
--- /dev/null
+++ b/relayer/chains/cosmos/msg_handlers_channel.go
@@ -0,0 +1,127 @@
+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"
+)
+
+// handleMsgChannelOpenInit will construct the start of the MsgChannelOpenTry
+// for the counterparty chain. PathProcessor will determine if this is needed.
+// For example, if a MsgChannelOpenTry or MsgChannelOpenConfirm is not detected on
+// the counterparty chain, and a MsgChannelOpenAck is not detected yet on this chain,
+// a MsgChannelOpenTry will be sent to the counterparty chain using this information
+// with the channel open init proof from this chain added.
+func (ccp *CosmosChainProcessor) handleMsgChannelOpenInit(p msgHandlerParams) bool {
+	ci := p.messageInfo.(*channelInfo)
+	k := ci.channelKey()
+	p.ibcMessagesCache.ChannelHandshake.Retain(k, processor.MsgChannelOpenInit, cosmos.NewCosmosMessage(&chantypes.MsgChannelOpenTry{
+		PortId:            k.PortID,
+		PreviousChannelId: k.ChannelID,
+		Channel: chantypes.Channel{
+			Counterparty: chantypes.Counterparty{
+				PortId:    k.CounterpartyPortID,
+				ChannelId: k.CounterpartyChannelID,
+			},
+			ConnectionHops: []string{ci.connectionID},
+		},
+	}))
+	// Setting false for open state because channel is not open until MsgChannelOpenAck on this chain.
+	ccp.channelStateCache[k] = false
+	ccp.logChannelMessage("MsgChannelOpenInit", ci)
+	return true
+}
+
+// handleMsgChannelOpenTry will construct the start of the MsgChannelOpenAck
+// for the counterparty chain. PathProcessor will determine if this is needed.
+// For example, if a MsgChannelOpenAck is not detected on the counterparty chain, and
+// a MsgChannelOpenConfirm is not detected yet on this chain, a MsgChannelOpenAck
+// will be sent to the counterparty chain using this information with the
+// channel open try proof from this chain added.
+func (ccp *CosmosChainProcessor) handleMsgChannelOpenTry(p msgHandlerParams) bool {
+	ci := p.messageInfo.(*channelInfo)
+	// using flipped counterparty since counterparty initialized this handshake
+	k := ci.channelKey().Counterparty()
+	p.ibcMessagesCache.ChannelHandshake.Retain(k, processor.MsgChannelOpenTry, cosmos.NewCosmosMessage(&chantypes.MsgChannelOpenAck{
+		PortId:                k.PortID,
+		ChannelId:             k.ChannelID,
+		CounterpartyChannelId: k.CounterpartyChannelID,
+	}))
+	// Setting false for open state because channel is not open until MsgChannelOpenConfirm on this chain.
+	ccp.channelStateCache[k] = false
+	ccp.logChannelMessage("MsgChannelOpenTry", ci)
+	return true
+}
+
+// handleMsgChannelOpenAck will construct the start of the MsgChannelOpenConfirm
+// for the counterparty chain. PathProcessor will determine if this is needed.
+// For example, if a MsgChannelOpenConfirm is not detected on the counterparty chain,
+// a MsgChannelOpenConfirm will be sent to the counterparty chain
+// using this information with the channel open ack proof from this chain added.
+func (ccp *CosmosChainProcessor) handleMsgChannelOpenAck(p msgHandlerParams) bool {
+	ci := p.messageInfo.(*channelInfo)
+	k := ci.channelKey()
+	p.ibcMessagesCache.ChannelHandshake.Retain(k, processor.MsgChannelOpenAck, cosmos.NewCosmosMessage(&chantypes.MsgChannelOpenConfirm{
+		PortId:    k.PortID,
+		ChannelId: k.ChannelID,
+	}))
+	ccp.channelStateCache[k] = true
+	ccp.logChannelMessage("MsgChannelOpenAck", ci)
+	return true
+}
+
+// handleMsgChannelOpenConfirm will retain 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 MsgChannelOpenConfirm is observed, but we want to
+// tell the PathProcessor that the channel handshake is complete for this channel.
+func (ccp *CosmosChainProcessor) handleMsgChannelOpenConfirm(p msgHandlerParams) bool {
+	ci := p.messageInfo.(*channelInfo)
+	// using flipped counterparty since counterparty initialized this handshake
+	k := ci.channelKey().Counterparty()
+	p.ibcMessagesCache.ChannelHandshake.Retain(k, processor.MsgChannelOpenConfirm, nil)
+	ccp.channelStateCache[k] = true
+	ccp.logChannelMessage("MsgChannelOpenConfirm", ci)
+	return true
+}
+
+// handleMsgChannelCloseInit will construct the start of the MsgChannelCloseConfirm
+// for the counterparty chain. PathProcessor will determine if this is needed.
+// For example, if a MsgChannelCloseConfirm is not detected on the counterparty chain,
+// a MsgChannelCloseConfirm will be sent to the counterparty chain
+// using this information with the channel close init proof from this chain added.
+func (ccp *CosmosChainProcessor) handleMsgChannelCloseInit(p msgHandlerParams) bool {
+	ci := p.messageInfo.(*channelInfo)
+	k := ci.channelKey()
+	p.ibcMessagesCache.ChannelHandshake.Retain(k, processor.MsgChannelCloseInit, cosmos.NewCosmosMessage(&chantypes.MsgChannelCloseConfirm{
+		PortId:    k.PortID,
+		ChannelId: k.ChannelID,
+	}))
+	ccp.channelStateCache[k] = false
+	ccp.logChannelMessage("MsgChannelCloseInit", ci)
+	return true
+}
+
+// handleMsgChannelCloseConfirm will retain a nil message 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 MsgChannelCloseConfirm is observed, but we want to
+// tell the PathProcessor that the channel close is complete for this channel.
+func (ccp *CosmosChainProcessor) handleMsgChannelCloseConfirm(p msgHandlerParams) bool {
+	ci := p.messageInfo.(*channelInfo)
+	// using flipped counterparty since counterparty initialized this channel close
+	k := ci.channelKey().Counterparty()
+	p.ibcMessagesCache.ChannelHandshake.Retain(k, processor.MsgChannelCloseConfirm, nil)
+	ccp.channelStateCache[k] = false
+	ccp.logChannelMessage("MsgChannelCloseConfirm", ci)
+	return true
+}
+
+func (ccp *CosmosChainProcessor) logChannelMessage(message string, channelInfo *channelInfo) {
+	ccp.logObservedIBCMessage(message,
+		zap.String("channel_id", channelInfo.channelID),
+		zap.String("port_id", channelInfo.portID),
+		zap.String("counterparty_channel_id", channelInfo.counterpartyChannelID),
+		zap.String("counterparty_port_id", channelInfo.counterpartyPortID),
+		zap.String("connection_id", channelInfo.connectionID),
+	)
+}


Test patch:
diff --git a/relayer/chains/cosmos/msg_handlers_channel_test.go b/relayer/chains/cosmos/msg_handlers_channel_test.go
new file mode 100644
index 0000000..a83f72b
--- /dev/null
+++ b/relayer/chains/cosmos/msg_handlers_channel_test.go
@@ -0,0 +1,118 @@
+package cosmos
+
+import (
+	"testing"
+
+	"github.com/cosmos/relayer/v2/relayer/processor"
+	"github.com/stretchr/testify/require"
+)
+
+func TestHandleChannelHandshake(t *testing.T) {
+	const (
+		srcChannel = "channel-0"
+		dstChannel = "channel-1"
+		srcPort    = "transfer"
+		dstPort    = "transfer"
+	)
+
+	ccp := mockCosmosChainProcessor(t)
+
+	channelInfo := &channelInfo{
+		channelID:             srcChannel,
+		portID:                srcPort,
+		counterpartyChannelID: dstChannel,
+		counterpartyPortID:    dstPort,
+	}
+
+	ibcMessagesCache := processor.NewIBCMessagesCache()
+
+	ccp.handleMsgChannelOpenInit(msgHandlerParams{messageInfo: channelInfo, ibcMessagesCache: ibcMessagesCache})
+
+	channelKey := channelInfo.channelKey()
+
+	channelOpen, ok := ccp.channelStateCache[channelKey]
+	require.True(t, ok, "unable to find channel state for channel key")
+
+	require.False(t, channelOpen, "channel should not be marked open yet")
+
+	require.Len(t, ibcMessagesCache.ChannelHandshake, 1)
+
+	channelMessages, hasChannelKey := ibcMessagesCache.ChannelHandshake[channelKey]
+	require.True(t, hasChannelKey, "no messages cached for channel key")
+
+	_, hasChannelOpenInit := channelMessages[processor.MsgChannelOpenInit]
+	require.True(t, hasChannelOpenInit, "no messages cached for MsgChannelOpenInit")
+
+	ccp.handleMsgChannelOpenAck(msgHandlerParams{messageInfo: channelInfo, ibcMessagesCache: ibcMessagesCache})
+
+	channelOpen, ok = ccp.channelStateCache[channelKey]
+	require.True(t, ok, "unable to find channel state for channel key")
+
+	require.True(t, channelOpen, "channel should be marked open now")
+
+	channelMessages, hasChannelKey = ibcMessagesCache.ChannelHandshake[channelKey]
+	require.True(t, hasChannelKey, "no messages cached for channel key")
+
+	require.Len(t, channelMessages, 2)
+
+	_, hasChannelOpenInit = channelMessages[processor.MsgChannelOpenInit]
+	require.True(t, hasChannelOpenInit, "no messages cached for MsgChannelOpenInit")
+
+	_, hasChannelOpenAck := channelMessages[processor.MsgChannelOpenAck]
+	require.True(t, hasChannelOpenAck, "no messages cached for MsgChannelOpenAck")
+}
+
+func TestHandleChannelHandshakeCounterparty(t *testing.T) {
+	const (
+		srcChannel = "channel-0"
+		dstChannel = "channel-1"
+		srcPort    = "transfer"
+		dstPort    = "transfer"
+	)
+
+	ccp := mockCosmosChainProcessor(t)
+
+	channelInfo := &channelInfo{
+		channelID:             srcChannel,
+		portID:                srcPort,
+		counterpartyChannelID: dstChannel,
+		counterpartyPortID:    dstPort,
+	}
+
+	ibcMessagesCache := processor.NewIBCMessagesCache()
+
+	ccp.handleMsgChannelOpenTry(msgHandlerParams{messageInfo: channelInfo, ibcMessagesCache: ibcMessagesCache})
+
+	channelKey := channelInfo.channelKey().Counterparty()
+
+	channelOpen, ok := ccp.channelStateCache[channelKey]
+	require.True(t, ok, "unable to find channel state for channel key")
+
+	require.False(t, channelOpen, "channel should not be marked open yet")
+
+	require.Len(t, ibcMessagesCache.ChannelHandshake, 1)
+
+	channelMessages, hasChannelKey := ibcMessagesCache.ChannelHandshake[channelKey]
+	require.True(t, hasChannelKey, "no messages cached for channel key")
+
+	_, hasChannelOpenTry := channelMessages[processor.MsgChannelOpenTry]
+	require.True(t, hasChannelOpenTry, "no messages cached for MsgChannelOpenTry")
+
+	ccp.handleMsgChannelOpenConfirm(msgHandlerParams{messageInfo: channelInfo, ibcMessagesCache: ibcMessagesCache})
+
+	channelOpen, ok = ccp.channelStateCache[channelKey]
+	require.True(t, ok, "unable to find channel state for channel key")
+
+	require.True(t, channelOpen, "channel should be marked open now")
+
+	channelMessages, hasChannelKey = ibcMessagesCache.ChannelHandshake[channelKey]
+	require.True(t, hasChannelKey, "no messages cached for channel key")
+
+	require.Len(t, channelMessages, 2)
+
+	_, hasChannelOpenTry = channelMessages[processor.MsgChannelOpenTry]
+	require.True(t, hasChannelOpenTry, "no messages cached for MsgChannelOpenTry")
+
+	_, hasChannelOpenConfirm := channelMessages[processor.MsgChannelOpenConfirm]
+	require.True(t, hasChannelOpenConfirm, "no messages cached for MsgChannelOpenConfirm")
+}