Handle ordered packets in UnreceivedPackets query.

CHANGELOG.md

@@ -8,6 +8,7 @@
 ### Improvements
 
 ### Bug Fixes
+* [\#15](https://github.com/Finschia/ibc-go/pull/15) Handle ordered packets in UnreceivedPackets query.
 
 ### Breaking Changes
 

modules/core/04-channel/keeper/grpc_query.go

@@ -397,18 +397,55 @@ func (q Keeper) UnreceivedPackets(c context.Context, req *types.QueryUnreceivedP
 
 	ctx := sdk.UnwrapSDKContext(c)
 
-	unreceivedSequences := []uint64{}
+	channel, found := q.GetChannel(sdk.UnwrapSDKContext(c), req.PortId, req.ChannelId)
+	if !found {
+		return nil, status.Error(
+			codes.NotFound,
+			sdkerrors.Wrapf(types.ErrChannelNotFound, "port-id: %s, channel-id %s", req.PortId, req.ChannelId).Error(),
+		)
+	}
 
-	for i, seq := range req.PacketCommitmentSequences {
-		if seq == 0 {
-			return nil, status.Errorf(codes.InvalidArgument, "packet sequence %d cannot be 0", i)
-		}
+	var unreceivedSequences []uint64
+	switch channel.Ordering {
+	case types.UNORDERED:
+		for i, seq := range req.PacketCommitmentSequences {
+			// filter for invalid sequences to ensure they are not included in the response value.
+			if seq == 0 {
+				return nil, status.Errorf(codes.InvalidArgument, "packet sequence %d cannot be 0", i)
+			}
 
-		// if packet receipt exists on the receiving chain, then packet has already been received
-		if _, found := q.GetPacketReceipt(ctx, req.PortId, req.ChannelId, seq); !found {
-			unreceivedSequences = append(unreceivedSequences, seq)
+			// if the packet receipt does not exist, then it is unreceived
+			if _, found := q.GetPacketReceipt(ctx, req.PortId, req.ChannelId, seq); !found {
+				unreceivedSequences = append(unreceivedSequences, seq)
+			}
+		}
+	case types.ORDERED:
+		nextSequenceRecv, found := q.GetNextSequenceRecv(ctx, req.PortId, req.ChannelId)
+		if !found {
+			return nil, status.Error(
+				codes.NotFound,
+				sdkerrors.Wrapf(
+					types.ErrSequenceReceiveNotFound,
+					"destination port: %s, destination channel: %s", req.PortId, req.ChannelId,
+				).Error(),
+			)
 		}
 
+		for i, seq := range req.PacketCommitmentSequences {
+			// filter for invalid sequences to ensure they are not included in the response value.
+			if seq == 0 {
+				return nil, status.Errorf(codes.InvalidArgument, "packet sequence %d cannot be 0", i)
+			}
+
+			// Any sequence greater than or equal to the next sequence to be received is not received.
+			if seq >= nextSequenceRecv {
+				unreceivedSequences = append(unreceivedSequences, seq)
+			}
+		}
+	default:
+		return nil, status.Error(
+			codes.InvalidArgument,
+			sdkerrors.Wrapf(types.ErrInvalidChannelOrdering, "channel order %s is not supported", channel.Ordering.String()).Error())
 	}
 
 	selfHeight := clienttypes.GetSelfHeight(ctx)

modules/core/04-channel/keeper/grpc_query_test.go

@@ -9,6 +9,7 @@ import (
 	clienttypes "github.com/Finschia/ibc-go/v3/modules/core/02-client/types"
 	connectiontypes "github.com/Finschia/ibc-go/v3/modules/core/03-connection/types"
 	"github.com/Finschia/ibc-go/v3/modules/core/04-channel/types"
+	ibchost "github.com/Finschia/ibc-go/v3/modules/core/24-host"
 	"github.com/Finschia/ibc-go/v3/modules/core/exported"
 	ibctesting "github.com/Finschia/ibc-go/v3/testing"
 )
@@ -1110,7 +1111,7 @@ func (suite *KeeperTestSuite) TestQueryPacketAcknowledgements() {
 func (suite *KeeperTestSuite) TestQueryUnreceivedPackets() {
 	var (
 		req    *types.QueryUnreceivedPacketsRequest
-		expSeq = []uint64{}
+		expSeq = []uint64(nil)
 	)
 
 	testCases := []struct {
@@ -1148,14 +1149,104 @@ func (suite *KeeperTestSuite) TestQueryUnreceivedPackets() {
 		{
 			"invalid seq",
 			func() {
+				path := ibctesting.NewPath(suite.chainA, suite.chainB)
+				suite.coordinator.Setup(path)
+
+				req = &types.QueryUnreceivedPacketsRequest{
+					PortId:                    path.EndpointA.ChannelConfig.PortID,
+					ChannelId:                 path.EndpointA.ChannelID,
+					PacketCommitmentSequences: []uint64{0},
+				}
+			},
+			false,
+		},
+		{
+			"invalid seq, ordered channel",
+			func() {
+				path := ibctesting.NewPath(suite.chainA, suite.chainB)
+				path.SetChannelOrdered()
+				suite.coordinator.Setup(path)
+
 				req = &types.QueryUnreceivedPacketsRequest{
-					PortId:                    "test-port-id",
-					ChannelId:                 "test-channel-id",
+					PortId:                    path.EndpointA.ChannelConfig.PortID,
+					ChannelId:                 path.EndpointA.ChannelID,
 					PacketCommitmentSequences: []uint64{0},
 				}
 			},
 			false,
 		},
+		{
+			"channel not found",
+			func() {
+				req = &types.QueryUnreceivedPacketsRequest{
+					PortId:    "invalid-port-id",
+					ChannelId: "invalid-channel-id",
+				}
+			},
+			false,
+		},
+		{
+			"invalid channel ordering",
+			func() {
+				path := ibctesting.NewPath(suite.chainA, suite.chainB)
+				suite.coordinator.Setup(path)
+
+				chKeepr := suite.chainA.App.GetIBCKeeper().ChannelKeeper
+				portID := path.EndpointA.ChannelConfig.PortID
+				chID := path.EndpointA.ChannelID
+				ctx := suite.chainA.GetContext()
+
+				ch, ok := chKeepr.GetChannel(ctx, portID, chID)
+				suite.Require().True(ok)
+				ch.Ordering = types.NONE
+				chKeepr.SetChannel(ctx, portID, chID, ch)
+
+				expSeq = []uint64{1}
+				req = &types.QueryUnreceivedPacketsRequest{
+					PortId:                    path.EndpointA.ChannelConfig.PortID,
+					ChannelId:                 path.EndpointA.ChannelID,
+					PacketCommitmentSequences: []uint64{1},
+				}
+			},
+			false,
+		},
+		{
+			"sequence receive not found",
+			func() {
+				path := ibctesting.NewPath(suite.chainA, suite.chainB)
+				path.SetChannelOrdered()
+				suite.coordinator.Setup(path)
+
+				portID := path.EndpointA.ChannelConfig.PortID
+				chID := path.EndpointA.ChannelID
+				ctx := suite.chainA.GetContext()
+				storeKey := suite.chainA.GetSimApp().GetKey(ibchost.StoreKey)
+				ctx.KVStore(storeKey).Delete(ibchost.NextSequenceRecvKey(portID, chID))
+
+				expSeq = []uint64(nil)
+				req = &types.QueryUnreceivedPacketsRequest{
+					PortId:                    path.EndpointA.ChannelConfig.PortID,
+					ChannelId:                 path.EndpointA.ChannelID,
+					PacketCommitmentSequences: []uint64{},
+				}
+			},
+			false,
+		},
+		{
+			"basic success empty packet commitments",
+			func() {
+				path := ibctesting.NewPath(suite.chainA, suite.chainB)
+				suite.coordinator.Setup(path)
+
+				expSeq = []uint64(nil)
+				req = &types.QueryUnreceivedPacketsRequest{
+					PortId:                    path.EndpointA.ChannelConfig.PortID,
+					ChannelId:                 path.EndpointA.ChannelID,
+					PacketCommitmentSequences: []uint64{},
+				}
+			},
+			true,
+		},
 		{
 			"basic success unreceived packet commitments",
 			func() {
@@ -1181,7 +1272,7 @@ func (suite *KeeperTestSuite) TestQueryUnreceivedPackets() {
 
 				suite.chainA.App.GetIBCKeeper().ChannelKeeper.SetPacketReceipt(suite.chainA.GetContext(), path.EndpointA.ChannelConfig.PortID, path.EndpointA.ChannelID, 1)
 
-				expSeq = []uint64{}
+				expSeq = []uint64(nil)
 				req = &types.QueryUnreceivedPacketsRequest{
 					PortId:                    path.EndpointA.ChannelConfig.PortID,
 					ChannelId:                 path.EndpointA.ChannelID,
@@ -1195,7 +1286,7 @@ func (suite *KeeperTestSuite) TestQueryUnreceivedPackets() {
 			func() {
 				path := ibctesting.NewPath(suite.chainA, suite.chainB)
 				suite.coordinator.Setup(path)
-				expSeq = []uint64{} // reset
+				expSeq = []uint64(nil) // reset
 				packetCommitments := []uint64{}
 
 				// set packet receipt for every other sequence
@@ -1217,6 +1308,60 @@ func (suite *KeeperTestSuite) TestQueryUnreceivedPackets() {
 			},
 			true,
 		},
+		{
+			"basic success empty packet commitments, ordered channel",
+			func() {
+				path := ibctesting.NewPath(suite.chainA, suite.chainB)
+				path.SetChannelOrdered()
+				suite.coordinator.Setup(path)
+
+				expSeq = []uint64(nil)
+				req = &types.QueryUnreceivedPacketsRequest{
+					PortId:                    path.EndpointA.ChannelConfig.PortID,
+					ChannelId:                 path.EndpointA.ChannelID,
+					PacketCommitmentSequences: []uint64{},
+				}
+			},
+			true,
+		},
+		{
+			"basic success unreceived packet commitments, ordered channel",
+			func() {
+				path := ibctesting.NewPath(suite.chainA, suite.chainB)
+				path.SetChannelOrdered()
+				suite.coordinator.Setup(path)
+
+				// Note: NextSequenceRecv is set to 1 on channel creation.
+				expSeq = []uint64{1}
+				req = &types.QueryUnreceivedPacketsRequest{
+					PortId:                    path.EndpointA.ChannelConfig.PortID,
+					ChannelId:                 path.EndpointA.ChannelID,
+					PacketCommitmentSequences: []uint64{1},
+				}
+			},
+			true,
+		},
+		{
+			"basic success multiple unreceived packet commitments, ordered channel",
+			func() {
+				path := ibctesting.NewPath(suite.chainA, suite.chainB)
+				path.SetChannelOrdered()
+				suite.coordinator.Setup(path)
+
+				// Exercise scenario from issue #1532. NextSequenceRecv is 5, packet commitments provided are 2, 7, 9, 10.
+				// Packet sequence 2 is already received so only sequences 7, 9, 10 should be considered unreceived.
+				expSeq = []uint64{7, 9, 10}
+				packetCommitments := []uint64{2, 7, 9, 10}
+				suite.chainA.App.GetIBCKeeper().ChannelKeeper.SetNextSequenceRecv(suite.chainA.GetContext(), path.EndpointA.ChannelConfig.PortID, path.EndpointA.ChannelID, 5)
+
+				req = &types.QueryUnreceivedPacketsRequest{
+					PortId:                    path.EndpointA.ChannelConfig.PortID,
+					ChannelId:                 path.EndpointA.ChannelID,
+					PacketCommitmentSequences: packetCommitments,
+				}
+			},
+			true,
+		},
 	}
 
 	for _, tc := range testCases {
@@ -1280,9 +1425,12 @@ func (suite *KeeperTestSuite) TestQueryUnreceivedAcks() {
 		{
 			"invalid seq",
 			func() {
+				path := ibctesting.NewPath(suite.chainA, suite.chainB)
+				suite.coordinator.Setup(path)
+
 				req = &types.QueryUnreceivedAcksRequest{
-					PortId:             "test-port-id",
-					ChannelId:          "test-channel-id",
+					PortId:             path.EndpointA.ChannelConfig.PortID,
+					ChannelId:          path.EndpointA.ChannelID,
 					PacketAckSequences: []uint64{0},
 				}
 			},

modules/core/keeper/msg_server.go

@@ -410,12 +410,11 @@ func (k Keeper) RecvPacket(goCtx context.Context, msg *channeltypes.MsgRecvPacke
 	// Cache context so that we may discard state changes from callback if the acknowledgement is unsuccessful.
 	cacheCtx, writeFn = ctx.CacheContext()
 	ack := cbs.OnRecvPacket(cacheCtx, msg.Packet, relayer)
-	// NOTE: The context returned by CacheContext() refers to a new EventManager, so it needs to explicitly set events to the original context.
-	// Events from callback are emitted regardless of acknowledgement success
-	ctx.EventManager().EmitEvents(cacheCtx.EventManager().Events())
 	if ack == nil || ack.Success() {
 		// write application state changes for asynchronous and successful acknowledgements
 		writeFn()
+		// NOTE: The context returned by CacheContext() refers to a new EventManager, so it needs to explicitly set events to the original context.
+		ctx.EventManager().EmitEvents(cacheCtx.EventManager().Events())
 	}
 
 	// Set packet acknowledgement only if the acknowledgement is not nil.