Repository: cosmos/relayer
Pull request #798 (2022-06-15 16:33:04)

CosmosChainProcessor Connection message handlers
Similar to `channelStateCache`, maintains `connectionStateCache` which will be used by the `PathProcessor` to determine if it can proceed with channel handshakes on a given connection.

Retains connection messages so that the `PathProcessor` will be able to correlate and execute relevant connection handshakes between clients.


Patch:
diff --git a/relayer/chains/cosmos/cosmos_chain_processor.go b/relayer/chains/cosmos/cosmos_chain_processor.go
index 34e51ec..2ea4668 100644
--- a/relayer/chains/cosmos/cosmos_chain_processor.go
+++ b/relayer/chains/cosmos/cosmos_chain_processor.go
@@ -31,6 +31,9 @@ type CosmosChainProcessor struct {
 	// holds highest consensus height and header for all clients
 	latestClientState
 
+	// holds open state for known connections
+	connectionStateCache processor.ConnectionStateCache
+
 	// holds open state for known channels
 	channelStateCache processor.ChannelStateCache
 }
@@ -41,12 +44,13 @@ 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,
-		latestClientState: make(latestClientState),
-		channelStateCache: make(processor.ChannelStateCache),
+		log:                  log.With(zap.String("chain_name", provider.ChainName()), zap.String("chain_id", provider.ChainId())),
+		chainProvider:        provider,
+		cc:                   cc,
+		pathProcessors:       pathProcessors,
+		latestClientState:    make(latestClientState),
+		connectionStateCache: make(processor.ConnectionStateCache),
+		channelStateCache:    make(processor.ChannelStateCache),
 	}, nil
 }
 
@@ -78,6 +82,11 @@ var messageHandlers = map[string]func(*CosmosChainProcessor, msgHandlerParams) b
 	processor.MsgUpdateClient:       (*CosmosChainProcessor).handleMsgUpdateClient,
 	processor.MsgUpgradeClient:      (*CosmosChainProcessor).handleMsgUpgradeClient,
 	processor.MsgSubmitMisbehaviour: (*CosmosChainProcessor).handleMsgSubmitMisbehaviour,
+
+	processor.MsgConnectionOpenInit:    (*CosmosChainProcessor).handleMsgConnectionOpenInit,
+	processor.MsgConnectionOpenTry:     (*CosmosChainProcessor).handleMsgConnectionOpenTry,
+	processor.MsgConnectionOpenAck:     (*CosmosChainProcessor).handleMsgConnectionOpenAck,
+	processor.MsgConnectionOpenConfirm: (*CosmosChainProcessor).handleMsgConnectionOpenConfirm,
 }
 
 func (ccp *CosmosChainProcessor) logObservedIBCMessage(m string, fields ...zap.Field) {
@@ -256,9 +265,10 @@ func (ccp *CosmosChainProcessor) queryCycle(ctx context.Context, persistence *qu
 
 	for _, pp := range ccp.pathProcessors {
 		pp.HandleNewData(chainID, processor.ChainProcessorCacheData{
-			IBCMessagesCache:  ibcMessagesCache,
-			InSync:            ccp.inSync,
-			ChannelStateCache: ccp.channelStateCache,
+			IBCMessagesCache:     ibcMessagesCache,
+			InSync:               ccp.inSync,
+			ConnectionStateCache: ccp.connectionStateCache,
+			ChannelStateCache:    ccp.channelStateCache,
 		})
 	}
 
diff --git a/relayer/chains/cosmos/msg_handlers_connection.go b/relayer/chains/cosmos/msg_handlers_connection.go
new file mode 100644
index 0000000..fc6f8d1
--- /dev/null
+++ b/relayer/chains/cosmos/msg_handlers_connection.go
@@ -0,0 +1,84 @@
+package cosmos
+
+import (
+	conntypes "github.com/cosmos/ibc-go/v3/modules/core/03-connection/types"
+	"github.com/cosmos/relayer/v2/relayer/processor"
+	"github.com/cosmos/relayer/v2/relayer/provider/cosmos"
+	"go.uber.org/zap"
+)
+
+// handleMsgConnectionOpenInit will construct the start of the MsgConnectionOpenTry
+// for the counterparty chain. PathProcessor will determine if this is needed.
+// For example, if a MsgConnectionOpenTry or MsgConnectionOpenConfirm is
+// not detected on the counterparty chain, and a MsgConnectionOpenAck is not detected
+// yet on this chain, a MsgConnectionOpenTry will be sent to the counterparty chain
+// using this information with the connection init proof from this chain added.
+func (ccp *CosmosChainProcessor) handleMsgConnectionOpenInit(p msgHandlerParams) bool {
+	ci := p.messageInfo.(*connectionInfo)
+	k := ci.connectionKey()
+	p.ibcMessagesCache.ConnectionHandshake.Retain(k, processor.MsgConnectionOpenInit, cosmos.NewCosmosMessage(&conntypes.MsgConnectionOpenTry{
+		ClientId:             k.ClientID,
+		PreviousConnectionId: k.ConnectionID,
+		Counterparty: conntypes.Counterparty{
+			ClientId:     k.CounterpartyClientID,
+			ConnectionId: k.CounterpartyConnectionID,
+		},
+	}))
+	ccp.connectionStateCache[k] = false
+	ccp.logConnectionMessage("MsgConnectionOpenInit", ci)
+	return true
+}
+
+// handleMsgConnectionOpenTry will construct the start of the MsgConnectionOpenAck
+// for the counterparty chain. PathProcessor will determine if this is needed.
+// For example, if a MsgConnectionOpenAck is not detected on the counterparty chain,
+// and a MsgConnectionOpenConfirm is not detected yet on this chain,
+// a MsgConnectionOpenAck will be sent to the counterparty chain
+// using this information with the connection try proof from this chain added.
+func (ccp *CosmosChainProcessor) handleMsgConnectionOpenTry(p msgHandlerParams) bool {
+	ci := p.messageInfo.(*connectionInfo)
+	k := ci.connectionKey().Counterparty()
+	p.ibcMessagesCache.ConnectionHandshake.Retain(k, processor.MsgConnectionOpenTry, cosmos.NewCosmosMessage(&conntypes.MsgConnectionOpenAck{
+		ConnectionId:             k.ConnectionID,
+		CounterpartyConnectionId: k.CounterpartyConnectionID,
+	}))
+	ccp.connectionStateCache[k] = false
+	ccp.logConnectionMessage("MsgConnectionOpenTry", ci)
+	return true
+}
+
+// handleMsgConnectionOpenAck will construct the start of the MsgConnectionOpenConfirm
+// for the counterparty chain. PathProcessor will determine if this is needed.
+// For example, if a MsgConnectionOpenConfirm is not detected on the counterparty chain,
+// a MsgConnectionOpenConfirm will be sent to the counterparty chain
+// using this information with the connection ack proof from this chain added.
+func (ccp *CosmosChainProcessor) handleMsgConnectionOpenAck(p msgHandlerParams) bool {
+	ci := p.messageInfo.(*connectionInfo)
+	k := ci.connectionKey()
+	p.ibcMessagesCache.ConnectionHandshake.Retain(k, processor.MsgConnectionOpenAck, cosmos.NewCosmosMessage(&conntypes.MsgConnectionOpenConfirm{
+		ConnectionId: k.ConnectionID,
+	}))
+	ccp.connectionStateCache[k] = true
+	ccp.logConnectionMessage("MsgConnectionOpenAck", ci)
+	return true
+}
+
+// handleMsgConnectionOpenConfirm will retaining a nil message here.
+// This is for book-keeping in the PathProcessor cache only.
+// A message does not need to be constructed for the counterparty chain after the MsgConnectionOpenConfirm is observed,
+// but we want to tell the PathProcessor that the connection handshake is complete for this sequence.
+func (ccp *CosmosChainProcessor) handleMsgConnectionOpenConfirm(p msgHandlerParams) bool {
+	ci := p.messageInfo.(*connectionInfo)
+	k := ci.connectionKey().Counterparty()
+	p.ibcMessagesCache.ConnectionHandshake.Retain(k, processor.MsgConnectionOpenConfirm, nil)
+	ccp.connectionStateCache[k] = true
+	ccp.logConnectionMessage("MsgConnectionOpenConfirm", ci)
+	return true
+}
+
+func (ccp *CosmosChainProcessor) logConnectionMessage(message string, connectionInfo *connectionInfo) {
+	ccp.logObservedIBCMessage(message, zap.String("client_id", connectionInfo.clientID),
+		zap.String("connection_id", connectionInfo.connectionID),
+		zap.String("counterparty_client_id", connectionInfo.counterpartyClientID),
+		zap.String("counterparty_connection_id", connectionInfo.counterpartyConnectionID))
+}
diff --git a/relayer/processor/path_processor.go b/relayer/processor/path_processor.go
index cc7b646..a73b79e 100644
--- a/relayer/processor/path_processor.go
+++ b/relayer/processor/path_processor.go
@@ -50,8 +50,9 @@ type PathEndRuntime struct {
 	chainProvider provider.ChainProvider
 
 	// cached data
-	messageCache      IBCMessagesCache
-	channelStateCache ChannelStateCache
+	messageCache         IBCMessagesCache
+	connectionStateCache ConnectionStateCache
+	channelStateCache    ChannelStateCache
 
 	// New messages and other data arriving from the handleNewMessagesForPathEnd method.
 	incomingCacheData chan ChainProcessorCacheData
@@ -62,8 +63,9 @@ type PathEndRuntime struct {
 
 func (pathEnd *PathEndRuntime) MergeCacheData(d ChainProcessorCacheData) {
 	// TODO make sure passes channel filter for pathEnd1 before calling this
-	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.messageCache.Merge(d.IBCMessagesCache)             // Merge incoming packet IBC messages into the backlog
+	pathEnd.connectionStateCache.Merge(d.ConnectionStateCache) // Update latest connection open state for chain
+	pathEnd.channelStateCache.Merge(d.ChannelStateCache)       // Update latest channel open state for chain
 	pathEnd.inSync = d.InSync
 }
 
@@ -78,16 +80,18 @@ func NewPathProcessor(log *zap.Logger, pathEnd1 PathEnd, pathEnd2 PathEnd) *Path
 	return &PathProcessor{
 		log: log,
 		pathEnd1: &PathEndRuntime{
-			info:              pathEnd1,
-			incomingCacheData: make(chan ChainProcessorCacheData, 100),
-			channelStateCache: make(ChannelStateCache),
-			messageCache:      NewIBCMessagesCache(),
+			info:                 pathEnd1,
+			incomingCacheData:    make(chan ChainProcessorCacheData, 100),
+			connectionStateCache: make(ConnectionStateCache),
+			channelStateCache:    make(ChannelStateCache),
+			messageCache:         NewIBCMessagesCache(),
 		},
 		pathEnd2: &PathEndRuntime{
-			info:              pathEnd2,
-			incomingCacheData: make(chan ChainProcessorCacheData, 100),
-			channelStateCache: make(ChannelStateCache),
-			messageCache:      NewIBCMessagesCache(),
+			info:                 pathEnd2,
+			incomingCacheData:    make(chan ChainProcessorCacheData, 100),
+			connectionStateCache: make(ConnectionStateCache),
+			channelStateCache:    make(ChannelStateCache),
+			messageCache:         NewIBCMessagesCache(),
 		},
 		retryProcess: make(chan struct{}, 8),
 	}


Test patch:
diff --git a/relayer/chains/cosmos/msg_handlers_connection_test.go b/relayer/chains/cosmos/msg_handlers_connection_test.go
new file mode 100644
index 0000000..c0f7602
--- /dev/null
+++ b/relayer/chains/cosmos/msg_handlers_connection_test.go
@@ -0,0 +1,118 @@
+package cosmos
+
+import (
+	"testing"
+
+	"github.com/cosmos/relayer/v2/relayer/processor"
+	"github.com/stretchr/testify/require"
+)
+
+func TestHandleConnectionHandshake(t *testing.T) {
+	const (
+		srcConnection = "connection-0"
+		dstConnection = "connection-1"
+		srcClient     = "client-0"
+		dstClient     = "client-1"
+	)
+
+	ccp := mockCosmosChainProcessor(t)
+
+	connectionInfo := &connectionInfo{
+		connectionID:             srcConnection,
+		clientID:                 srcClient,
+		counterpartyClientID:     dstClient,
+		counterpartyConnectionID: dstConnection,
+	}
+
+	ibcMessagesCache := processor.NewIBCMessagesCache()
+
+	ccp.handleMsgConnectionOpenInit(msgHandlerParams{messageInfo: connectionInfo, ibcMessagesCache: ibcMessagesCache})
+
+	connectionKey := connectionInfo.connectionKey()
+
+	connectionOpen, ok := ccp.connectionStateCache[connectionKey]
+	require.True(t, ok, "unable to find connection state for connection key")
+
+	require.False(t, connectionOpen, "connection should not be marked open yet")
+
+	require.Len(t, ibcMessagesCache.ConnectionHandshake, 1)
+
+	connectionMessages, hasConnectionKey := ibcMessagesCache.ConnectionHandshake[connectionKey]
+	require.True(t, hasConnectionKey, "no messages cached for connection key")
+
+	_, hasConnectionOpenInit := connectionMessages[processor.MsgConnectionOpenInit]
+	require.True(t, hasConnectionOpenInit, "no messages cached for MsgConnectionOpenInit")
+
+	ccp.handleMsgConnectionOpenAck(msgHandlerParams{messageInfo: connectionInfo, ibcMessagesCache: ibcMessagesCache})
+
+	connectionOpen, ok = ccp.connectionStateCache[connectionKey]
+	require.True(t, ok, "unable to find connection state for connection key")
+
+	require.True(t, connectionOpen, "connection should be marked open now")
+
+	connectionMessages, hasConnectionKey = ibcMessagesCache.ConnectionHandshake[connectionKey]
+	require.True(t, hasConnectionKey, "no messages cached for connection key")
+
+	require.Len(t, connectionMessages, 2)
+
+	_, hasConnectionOpenInit = connectionMessages[processor.MsgConnectionOpenInit]
+	require.True(t, hasConnectionOpenInit, "no messages cached for MsgConnectionOpenInit")
+
+	_, hasConnectionOpenAck := connectionMessages[processor.MsgConnectionOpenAck]
+	require.True(t, hasConnectionOpenAck, "no messages cached for MsgConnectionOpenAck")
+}
+
+func TestHandleConnectionHandshakeCounterparty(t *testing.T) {
+	const (
+		srcConnection = "connection-0"
+		dstConnection = "connection-1"
+		srcClient     = "client-0"
+		dstClient     = "client-1"
+	)
+
+	ccp := mockCosmosChainProcessor(t)
+
+	connectionInfo := &connectionInfo{
+		connectionID:             srcConnection,
+		clientID:                 srcClient,
+		counterpartyClientID:     dstClient,
+		counterpartyConnectionID: dstConnection,
+	}
+
+	ibcMessagesCache := processor.NewIBCMessagesCache()
+
+	ccp.handleMsgConnectionOpenTry(msgHandlerParams{messageInfo: connectionInfo, ibcMessagesCache: ibcMessagesCache})
+
+	connectionKey := connectionInfo.connectionKey().Counterparty()
+
+	connectionOpen, ok := ccp.connectionStateCache[connectionKey]
+	require.True(t, ok, "unable to find connection state for connection key")
+
+	require.False(t, connectionOpen, "connection should not be marked open yet")
+
+	require.Len(t, ibcMessagesCache.ConnectionHandshake, 1)
+
+	connectionMessages, hasConnectionKey := ibcMessagesCache.ConnectionHandshake[connectionKey]
+	require.True(t, hasConnectionKey, "no messages cached for connection key")
+
+	_, hasConnectionOpenTry := connectionMessages[processor.MsgConnectionOpenTry]
+	require.True(t, hasConnectionOpenTry, "no messages cached for MsgConnectionOpenTry")
+
+	ccp.handleMsgConnectionOpenConfirm(msgHandlerParams{messageInfo: connectionInfo, ibcMessagesCache: ibcMessagesCache})
+
+	connectionOpen, ok = ccp.connectionStateCache[connectionKey]
+	require.True(t, ok, "unable to find connection state for connection key")
+
+	require.True(t, connectionOpen, "connection should be marked open now")
+
+	connectionMessages, hasConnectionKey = ibcMessagesCache.ConnectionHandshake[connectionKey]
+	require.True(t, hasConnectionKey, "no messages cached for connection key")
+
+	require.Len(t, connectionMessages, 2)
+
+	_, hasConnectionOpenTry = connectionMessages[processor.MsgConnectionOpenTry]
+	require.True(t, hasConnectionOpenTry, "no messages cached for MsgConnectionOpenTry")
+
+	_, hasConnectionOpenConfirm := connectionMessages[processor.MsgConnectionOpenConfirm]
+	require.True(t, hasConnectionOpenConfirm, "no messages cached for MsgConnectionOpenConfirm")
+}