Repository: cosmos/relayer
Pull request #1038 (2022-11-06 18:57:23)

recv_packet/write_acknowledgement separation
Separate `recv_packet` from `write_acknowledgement` events since they can be in separate blocks with forward middleware.

This moves the duty of combining the data from the `write_acknowledgement` event with the `recv_packet` event into the path processor, which simplifies the event parser.


Patch:
diff --git a/relayer/chains/cosmos/event_parser.go b/relayer/chains/cosmos/event_parser.go
index 80650f1..ddc321d 100644
--- a/relayer/chains/cosmos/event_parser.go
+++ b/relayer/chains/cosmos/event_parser.go
@@ -49,43 +49,15 @@ func ibcMessagesFromEvents(
 	chainID string,
 	height uint64,
 ) (messages []ibcMessage) {
-	recvPacketMsgs := make(map[packetKey]*packetInfo)
 	for _, event := range events {
 		evt := sdk.StringifyEvent(event)
-		switch event.Type {
-		case chantypes.EventTypeRecvPacket, chantypes.EventTypeWriteAck:
-			pi := &packetInfo{Height: height}
-			pi.parseAttrs(log, evt.Attributes)
-			ck, err := processor.PacketInfoChannelKey(event.Type, provider.PacketInfo(*pi))
-			if err == nil {
-				pk := packetKey{
-					sequence: pi.Sequence,
-					channel:  ck,
-				}
-				_, ok := recvPacketMsgs[pk]
-				if !ok {
-					recvPacketMsgs[pk] = pi
-				} else {
-					recvPacketMsgs[pk].parseAttrs(log, evt.Attributes)
-				}
-			}
-		default:
-			m := parseIBCMessageFromEvent(log, evt, chainID, height)
-			if m == nil || m.info == nil {
-				// Not an IBC message, don't need to log here
-				continue
-			}
-			messages = append(messages, *m)
+		m := parseIBCMessageFromEvent(log, evt, chainID, height)
+		if m == nil || m.info == nil {
+			// Not an IBC message, don't need to log here
+			continue
 		}
+		messages = append(messages, *m)
 	}
-
-	for _, recvPacketMsg := range recvPacketMsgs {
-		messages = append(messages, ibcMessage{
-			eventType: chantypes.EventTypeRecvPacket,
-			info:      recvPacketMsg,
-		})
-	}
-
 	return messages
 }
 
@@ -96,7 +68,7 @@ func parseIBCMessageFromEvent(
 	height uint64,
 ) *ibcMessage {
 	switch event.Type {
-	case chantypes.EventTypeSendPacket,
+	case chantypes.EventTypeSendPacket, chantypes.EventTypeRecvPacket, chantypes.EventTypeWriteAck,
 		chantypes.EventTypeAcknowledgePacket, chantypes.EventTypeTimeoutPacket,
 		chantypes.EventTypeTimeoutPacketOnClose:
 		pi := &packetInfo{Height: height}
diff --git a/relayer/chains/cosmos/message_handlers.go b/relayer/chains/cosmos/message_handlers.go
index 254037a..8f86a7d 100644
--- a/relayer/chains/cosmos/message_handlers.go
+++ b/relayer/chains/cosmos/message_handlers.go
@@ -36,11 +36,6 @@ func (ccp *CosmosChainProcessor) handlePacketMessage(eventType string, pi provid
 		return
 	}
 
-	if eventType == chantypes.EventTypeRecvPacket && len(pi.Ack) == 0 {
-		// ignore recv packet with empty ack bytes
-		return
-	}
-
 	if !c.PacketFlow.ShouldRetainSequence(ccp.pathProcessors, k, ccp.chainProvider.ChainId(), eventType, pi.Sequence) {
 		ccp.log.Debug("Not retaining packet message",
 			zap.String("event_type", eventType),
diff --git a/relayer/chains/cosmos/query.go b/relayer/chains/cosmos/query.go
index 1f2cad1..20bacf2 100644
--- a/relayer/chains/cosmos/query.go
+++ b/relayer/chains/cosmos/query.go
@@ -716,8 +716,11 @@ func (cc *CosmosProvider) QuerySendPacket(
 		return provider.PacketInfo{}, err
 	}
 	for _, msg := range ibcMsgs {
+		if msg.eventType != chantypes.EventTypeSendPacket {
+			continue
+		}
 		if pi, ok := msg.info.(*packetInfo); ok {
-			if pi.Sequence == sequence {
+			if pi.SourceChannel == srcChanID && pi.SourcePort == srcPortID && pi.Sequence == sequence {
 				return provider.PacketInfo(*pi), nil
 			}
 		}
@@ -737,8 +740,11 @@ func (cc *CosmosProvider) QueryRecvPacket(
 		return provider.PacketInfo{}, err
 	}
 	for _, msg := range ibcMsgs {
+		if msg.eventType != chantypes.EventTypeWriteAck {
+			continue
+		}
 		if pi, ok := msg.info.(*packetInfo); ok {
-			if pi.DestChannel == dstChanID && pi.Sequence == sequence {
+			if pi.DestChannel == dstChanID && pi.DestPort == dstPortID && pi.Sequence == sequence {
 				return provider.PacketInfo(*pi), nil
 			}
 		}
diff --git a/relayer/processor/path_processor_internal.go b/relayer/processor/path_processor_internal.go
index e5eada1..3d1e884 100644
--- a/relayer/processor/path_processor_internal.go
+++ b/relayer/processor/path_processor_internal.go
@@ -52,6 +52,7 @@ MsgTransferLoop:
 				// remove all retention of this sequence number
 				res.ToDeleteSrc[chantypes.EventTypeSendPacket] = append(res.ToDeleteSrc[chantypes.EventTypeSendPacket], transferSeq)
 				res.ToDeleteDst[chantypes.EventTypeRecvPacket] = append(res.ToDeleteDst[chantypes.EventTypeRecvPacket], transferSeq)
+				res.ToDeleteDst[chantypes.EventTypeWriteAck] = append(res.ToDeleteDst[chantypes.EventTypeWriteAck], transferSeq)
 				res.ToDeleteSrc[chantypes.EventTypeAcknowledgePacket] = append(res.ToDeleteSrc[chantypes.EventTypeAcknowledgePacket], transferSeq)
 				continue MsgTransferLoop
 			}
@@ -106,6 +107,10 @@ MsgTransferLoop:
 		}
 		for msgRecvSeq, msgAcknowledgement := range pathEndPacketFlowMessages.DstMsgRecvPacket {
 			if transferSeq == msgRecvSeq {
+				if len(msgAcknowledgement.Ack) == 0 {
+					// have recv_packet but not write_acknowledgement yet. skip for now.
+					continue MsgTransferLoop
+				}
 				// msg is received by dst chain, but no ack yet. Need to relay ack from dst to src!
 				ackMsg := packetIBCMessage{
 					eventType: chantypes.EventTypeAcknowledgePacket,
@@ -179,6 +184,7 @@ MsgTransferLoop:
 	for ackSeq := range pathEndPacketFlowMessages.SrcMsgAcknowledgement {
 		res.ToDeleteSrc[chantypes.EventTypeSendPacket] = append(res.ToDeleteSrc[chantypes.EventTypeSendPacket], ackSeq)
 		res.ToDeleteDst[chantypes.EventTypeRecvPacket] = append(res.ToDeleteDst[chantypes.EventTypeRecvPacket], ackSeq)
+		res.ToDeleteDst[chantypes.EventTypeWriteAck] = append(res.ToDeleteDst[chantypes.EventTypeWriteAck], ackSeq)
 		res.ToDeleteSrc[chantypes.EventTypeAcknowledgePacket] = append(res.ToDeleteSrc[chantypes.EventTypeAcknowledgePacket], ackSeq)
 	}
 	for timeoutSeq, msgTimeout := range pathEndPacketFlowMessages.SrcMsgTimeout {
@@ -582,12 +588,29 @@ func (pp *PathProcessor) processLatestMessages(ctx context.Context, messageLifec
 			}
 		}
 
+		// Append acks into recv packet info if present
+		pathEnd1DstMsgRecvPacket := pp.pathEnd2.messageCache.PacketFlow[pair.pathEnd2ChannelKey][chantypes.EventTypeRecvPacket]
+		for seq, ackInfo := range pp.pathEnd2.messageCache.PacketFlow[pair.pathEnd2ChannelKey][chantypes.EventTypeWriteAck] {
+			if recvPacketInfo, ok := pathEnd1DstMsgRecvPacket[seq]; ok {
+				recvPacketInfo.Ack = ackInfo.Ack
+				pathEnd1DstMsgRecvPacket[seq] = recvPacketInfo
+			}
+		}
+
+		pathEnd2DstMsgRecvPacket := pp.pathEnd1.messageCache.PacketFlow[pair.pathEnd1ChannelKey][chantypes.EventTypeRecvPacket]
+		for seq, ackInfo := range pp.pathEnd1.messageCache.PacketFlow[pair.pathEnd1ChannelKey][chantypes.EventTypeWriteAck] {
+			if recvPacketInfo, ok := pathEnd2DstMsgRecvPacket[seq]; ok {
+				recvPacketInfo.Ack = ackInfo.Ack
+				pathEnd2DstMsgRecvPacket[seq] = recvPacketInfo
+			}
+		}
+
 		pathEnd1PacketFlowMessages := pathEndPacketFlowMessages{
 			Src:                       pp.pathEnd1,
 			Dst:                       pp.pathEnd2,
 			ChannelKey:                pair.pathEnd1ChannelKey,
 			SrcMsgTransfer:            pp.pathEnd1.messageCache.PacketFlow[pair.pathEnd1ChannelKey][chantypes.EventTypeSendPacket],
-			DstMsgRecvPacket:          pp.pathEnd2.messageCache.PacketFlow[pair.pathEnd2ChannelKey][chantypes.EventTypeRecvPacket],
+			DstMsgRecvPacket:          pathEnd1DstMsgRecvPacket,
 			SrcMsgAcknowledgement:     pp.pathEnd1.messageCache.PacketFlow[pair.pathEnd1ChannelKey][chantypes.EventTypeAcknowledgePacket],
 			SrcMsgTimeout:             pp.pathEnd1.messageCache.PacketFlow[pair.pathEnd1ChannelKey][chantypes.EventTypeTimeoutPacket],
 			SrcMsgTimeoutOnClose:      pp.pathEnd1.messageCache.PacketFlow[pair.pathEnd1ChannelKey][chantypes.EventTypeTimeoutPacketOnClose],
@@ -598,7 +621,7 @@ func (pp *PathProcessor) processLatestMessages(ctx context.Context, messageLifec
 			Dst:                       pp.pathEnd1,
 			ChannelKey:                pair.pathEnd2ChannelKey,
 			SrcMsgTransfer:            pp.pathEnd2.messageCache.PacketFlow[pair.pathEnd2ChannelKey][chantypes.EventTypeSendPacket],
-			DstMsgRecvPacket:          pp.pathEnd1.messageCache.PacketFlow[pair.pathEnd1ChannelKey][chantypes.EventTypeRecvPacket],
+			DstMsgRecvPacket:          pathEnd2DstMsgRecvPacket,
 			SrcMsgAcknowledgement:     pp.pathEnd2.messageCache.PacketFlow[pair.pathEnd2ChannelKey][chantypes.EventTypeAcknowledgePacket],
 			SrcMsgTimeout:             pp.pathEnd2.messageCache.PacketFlow[pair.pathEnd2ChannelKey][chantypes.EventTypeTimeoutPacket],
 			SrcMsgTimeoutOnClose:      pp.pathEnd2.messageCache.PacketFlow[pair.pathEnd2ChannelKey][chantypes.EventTypeTimeoutPacketOnClose],


Test patch:
diff --git a/relayer/chains/cosmos/event_parser_test.go b/relayer/chains/cosmos/event_parser_test.go
index 1cd823a..cab9ff6 100644
--- a/relayer/chains/cosmos/event_parser_test.go
+++ b/relayer/chains/cosmos/event_parser_test.go
@@ -302,7 +302,7 @@ func TestParseEventLogs(t *testing.T) {
 
 	ibcMessages := ibcMessagesFromEvents(zap.NewNop(), events, "", 0)
 
-	require.Len(t, ibcMessages, 2)
+	require.Len(t, ibcMessages, 3)
 
 	msgUpdateClient := ibcMessages[0]
 	require.Equal(t, clienttypes.EventTypeUpdateClient, msgUpdateClient.eventType)
@@ -322,7 +322,13 @@ func TestParseEventLogs(t *testing.T) {
 	require.Equal(t, chantypes.EventTypeRecvPacket, msgRecvPacket.eventType, "message event is not recv_packet")
 
 	packetInfoParsed, isPacketInfo := msgRecvPacket.info.(*packetInfo)
-	require.True(t, isPacketInfo, "messageInfo is not packetInfo")
+	require.True(t, isPacketInfo, "recv_packet messageInfo is not packetInfo")
+
+	msgWriteAcknowledgement := ibcMessages[2]
+	require.Equal(t, chantypes.EventTypeWriteAck, msgWriteAcknowledgement.eventType, "message event is not write_acknowledgement")
+
+	ackPacketInfoParsed, isPacketInfo := msgWriteAcknowledgement.info.(*packetInfo)
+	require.True(t, isPacketInfo, "ack messageInfo is not packetInfo")
 
 	packetAck, err := hex.DecodeString(testPacketAckHex)
 	require.NoError(t, err, "error decoding test packet ack")
@@ -342,6 +348,14 @@ func TestParseEventLogs(t *testing.T) {
 		SourcePort:       testPacketSrcPort,
 		DestChannel:      testPacketDstChannel,
 		DestPort:         testPacketDstPort,
-		Ack:              packetAck,
+	}), "parsed packet info does not match expected")
+
+	require.Empty(t, cmp.Diff(provider.PacketInfo(*ackPacketInfoParsed), provider.PacketInfo{
+		Sequence:      uint64(1),
+		SourceChannel: testPacketSrcChannel,
+		SourcePort:    testPacketSrcPort,
+		DestChannel:   testPacketDstChannel,
+		DestPort:      testPacketDstPort,
+		Ack:           packetAck,
 	}), "parsed packet info does not match expected")
 }