feat(delayedack): added efficient query for pending packets by addr (#…

…1385)

app/apptesting/delayedack.go

@@ -0,0 +1,53 @@
+package apptesting
+
+import (
+	"testing"
+
+	transfertypes "github.com/cosmos/ibc-go/v7/modules/apps/transfer/types"
+	channeltypes "github.com/cosmos/ibc-go/v7/modules/core/04-channel/types"
+	"github.com/stretchr/testify/require"
+
+	commontypes "github.com/dymensionxyz/dymension/v3/x/common/types"
+)
+
+const (
+	TestPacketReceiver = "testReceiver"
+	TestPacketSender   = "testSender"
+)
+
+func GenerateTestPacketData(t *testing.T) []byte {
+	t.Helper()
+	data := &transfertypes.FungibleTokenPacketData{
+		Receiver: TestPacketReceiver,
+		Sender:   TestPacketSender,
+	}
+	pd, err := transfertypes.ModuleCdc.MarshalJSON(data)
+	require.NoError(t, err)
+	return pd
+}
+
+func GenerateTestPacket(t *testing.T, sequence uint64) *channeltypes.Packet {
+	t.Helper()
+	return &channeltypes.Packet{
+		SourcePort:         "testSourcePort",
+		SourceChannel:      "testSourceChannel",
+		DestinationPort:    "testDestinationPort",
+		DestinationChannel: "testDestinationChannel",
+		Data:               GenerateTestPacketData(t),
+		Sequence:           sequence,
+	}
+}
+
+func GenerateRollappPackets(t *testing.T, rollappId string, num uint64) []commontypes.RollappPacket {
+	t.Helper()
+	var packets []commontypes.RollappPacket
+	for i := uint64(0); i < num; i++ {
+		packets = append(packets, commontypes.RollappPacket{
+			RollappId:   rollappId,
+			Packet:      GenerateTestPacket(t, i),
+			Status:      commontypes.Status_PENDING,
+			ProofHeight: i,
+		})
+	}
+	return packets
+}

app/apptesting/test_suite.go

@@ -183,13 +183,12 @@ func FundForAliasRegistration(
 	)
 }
 
-func (s *KeeperTestHelper) FinalizeAllPendingPackets(rollappID, receiver string) int {
+func (s *KeeperTestHelper) FinalizeAllPendingPackets(address string) int {
 	s.T().Helper()
-	// Query all pending packets by receiver
+	// Query all pending packets by address
 	querier := delayedackkeeper.NewQuerier(s.App.DelayedAckKeeper)
-	resp, err := querier.GetPendingPacketsByReceiver(s.Ctx, &delayedacktypes.QueryPendingPacketsByReceiverRequest{
-		RollappId: rollappID,
-		Receiver:  receiver,
+	resp, err := querier.GetPendingPacketsByAddress(s.Ctx, &delayedacktypes.QueryPendingPacketsByAddressRequest{
+		Address: address,
 	})
 	s.Require().NoError(err)
 	// Finalize all packets and return the num of finalized

app/upgrades/v4/upgrade.go

@@ -19,13 +19,13 @@ import (
 	upgradetypes "github.com/cosmos/cosmos-sdk/x/upgrade/types"
 	ibctransfertypes "github.com/cosmos/ibc-go/v7/modules/apps/transfer/types"
 	ibcchannelkeeper "github.com/cosmos/ibc-go/v7/modules/core/04-channel/keeper"
-
 	evmtypes "github.com/evmos/ethermint/x/evm/types"
 	feemarkettypes "github.com/evmos/ethermint/x/feemarket/types"
 	epochskeeper "github.com/osmosis-labs/osmosis/v15/x/epochs/keeper"
 
 	"github.com/dymensionxyz/dymension/v3/app/keepers"
 	"github.com/dymensionxyz/dymension/v3/app/upgrades"
+	commontypes "github.com/dymensionxyz/dymension/v3/x/common/types"
 	delayedackkeeper "github.com/dymensionxyz/dymension/v3/x/delayedack/keeper"
 	delayedacktypes "github.com/dymensionxyz/dymension/v3/x/delayedack/types"
 	incentiveskeeper "github.com/dymensionxyz/dymension/v3/x/incentives/keeper"
@@ -69,6 +69,10 @@ func CreateUpgradeHandler(
 			return nil, err
 		}
 
+		if err := migrateDelayedAckPacketIndex(ctx, keepers.DelayedAckKeeper); err != nil {
+			return nil, err
+		}
+
 		// Start running the module migrations
 		logger.Debug("running module migrations ...")
 		return mm.RunMigrations(ctx, configurator, fromVM)
@@ -215,6 +219,24 @@ func migrateIncentivesParams(ctx sdk.Context, ik *incentiveskeeper.Keeper) {
 	ik.SetParams(ctx, params)
 }
 
+func migrateDelayedAckPacketIndex(ctx sdk.Context, dk delayedackkeeper.Keeper) error {
+	pendingPackets := dk.ListRollappPackets(ctx, delayedacktypes.ByStatus(commontypes.Status_PENDING))
+	for _, packet := range pendingPackets {
+		pd, err := packet.GetTransferPacketData()
+		if err != nil {
+			return err
+		}
+
+		switch packet.Type {
+		case commontypes.RollappPacket_ON_RECV:
+			dk.MustSetPendingPacketByAddress(ctx, pd.Receiver, packet.RollappPacketKey())
+		case commontypes.RollappPacket_ON_ACK, commontypes.RollappPacket_ON_TIMEOUT:
+			dk.MustSetPendingPacketByAddress(ctx, pd.Sender, packet.RollappPacketKey())
+		}
+	}
+	return nil
+}
+
 func ConvertOldRollappToNew(oldRollapp rollapptypes.Rollapp) rollapptypes.Rollapp {
 	return rollapptypes.Rollapp{
 		RollappId:    oldRollapp.RollappId,

app/upgrades/v4/upgrade_test.go

@@ -19,6 +19,8 @@ import (
 	"github.com/dymensionxyz/dymension/v3/app/apptesting"
 	v4 "github.com/dymensionxyz/dymension/v3/app/upgrades/v4"
 	"github.com/dymensionxyz/dymension/v3/testutil/sample"
+	commontypes "github.com/dymensionxyz/dymension/v3/x/common/types"
+	delayedacktypes "github.com/dymensionxyz/dymension/v3/x/delayedack/types"
 	rollapptypes "github.com/dymensionxyz/dymension/v3/x/rollapp/types"
 	sequencertypes "github.com/dymensionxyz/dymension/v3/x/sequencer/types"
 	streamertypes "github.com/dymensionxyz/dymension/v3/x/streamer/types"
@@ -74,6 +76,8 @@ func (s *UpgradeTestSuite) TestUpgrade() {
 				// Create and store sequencers
 				s.seedAndStoreSequencers(numRollapps)
 
+				s.seedPendingRollappPackets()
+
 				return nil
 			},
 			upgrade: func() {
@@ -126,6 +130,11 @@ func (s *UpgradeTestSuite) TestUpgrade() {
 					return
 				}
 
+				// Check rollapp packets
+				if err = s.validateDelayedAckIndexMigration(); err != nil {
+					return
+				}
+
 				s.validateStreamerMigration()
 
 				return
@@ -282,6 +291,14 @@ func (s *UpgradeTestSuite) validateStreamerMigration() {
 	s.Require().Equal(expected, pointers)
 }
 
+func (s *UpgradeTestSuite) validateDelayedAckIndexMigration() error {
+	packets := s.App.DelayedAckKeeper.ListRollappPackets(s.Ctx, delayedacktypes.ByStatus(commontypes.Status_PENDING))
+	actual, err := s.App.DelayedAckKeeper.GetPendingPacketsByAddress(s.Ctx, apptesting.TestPacketReceiver)
+	s.Require().NoError(err)
+	s.Require().Equal(len(packets), len(actual))
+	return nil
+}
+
 func (s *UpgradeTestSuite) seedAndStoreRollapps(numRollapps int) {
 	for _, rollapp := range s.seedRollapps(numRollapps) {
 		s.App.RollappKeeper.SetRollapp(s.Ctx, rollapp)
@@ -340,3 +357,10 @@ func (s *UpgradeTestSuite) seedSequencers(numRollapps int) []sequencertypes.Sequ
 func rollappIDFromIdx(idx int) string {
 	return fmt.Sprintf("roll%spp_123%d-1", string(rune(idx+'a')), idx+1)
 }
+
+func (s *UpgradeTestSuite) seedPendingRollappPackets() {
+	packets := apptesting.GenerateRollappPackets(s.T(), "testrollappid_1-1", 20)
+	for _, packet := range packets {
+		s.App.DelayedAckKeeper.SetRollappPacket(s.Ctx, packet)
+	}
+}

ibctesting/bridging_fee_test.go

@@ -97,7 +97,7 @@ func (s *bridgingFeeSuite) TestBridgingFee() {
 	s.Require().NoError(err)
 
 	// manually finalize packets through x/delayedack
-	s.finalizeRollappPacketsByReceiver(s.hubChain().SenderAccount.GetAddress().String())
+	s.finalizeRollappPacketsByAddress(s.hubChain().SenderAccount.GetAddress().String())
 
 	// check balance after finalization
 	expectedFee := s.hubApp().DelayedAckKeeper.BridgingFeeFromAmt(s.hubCtx(), transferredCoins.Amount)

ibctesting/delayed_ack_test.go

@@ -174,7 +174,7 @@ func (s *delayedAckSuite) TestTransferRollappToHubFinalization() {
 	s.Require().NoError(err)
 
 	// manually finalize packets through x/delayedack
-	s.finalizeRollappPacketsByReceiver(s.hubChain().SenderAccount.GetAddress().String())
+	s.finalizeRollappPacketsByAddress(s.hubChain().SenderAccount.GetAddress().String())
 
 	// Validate ack is found
 	found = hubIBCKeeper.ChannelKeeper.HasPacketAcknowledgement(s.hubCtx(), packet.GetDestPort(), packet.GetDestChannel(), packet.GetSequence())
@@ -230,7 +230,7 @@ func (s *delayedAckSuite) TestHubToRollappTimeout() {
 	_, err = s.finalizeRollappState(1, currentRollappBlockHeight)
 	s.Require().NoError(err)
 	// manually finalize packets through x/delayedack
-	s.finalizeRollappPacketsByReceiver(receiverAccount.String())
+	s.finalizeRollappPacketsByAddress(senderAccount.String())
 	// Validate funds are returned to the sender
 	postFinalizeBalance := bankKeeper.GetBalance(s.hubCtx(), senderAccount, sdk.DefaultBondDenom)
 	s.Require().Equal(preSendBalance.Amount, postFinalizeBalance.Amount)

ibctesting/eibc_test.go

@@ -262,7 +262,7 @@ func (s *eibcSuite) TestEIBCDemandOrderFulfillment() {
 				s.Require().NoError(err)
 
 				// manually finalize packets through x/delayedack
-				s.finalizeRollappPacketsByReceiver(fulfiller.String())
+				s.finalizeRollappPacketsByAddress(fulfiller.String())
 
 				// Check the fulfiller balance was updated fully with the IBC amount
 				isUpdated := false
@@ -346,7 +346,7 @@ func (s *eibcSuite) TestEIBCDemandOrderFulfillment() {
 			s.Require().NoError(err)
 
 			// manually finalize packets through x/delayedack
-			evts := s.finalizeRollappPacketsByReceiver(fulfiller.String())
+			evts := s.finalizeRollappPacketsByAddress(fulfiller.String())
 
 			ack, err := ibctesting.ParseAckFromEvents(evts)
 			s.Require().NoError(err)
@@ -509,7 +509,7 @@ func (s *eibcSuite) TestTimeoutEIBCDemandOrderFulfillment() {
 			_, err = s.finalizeRollappState(1, currentRollappBlockHeight)
 			s.Require().NoError(err)
 			// manually finalize packets through x/delayedack
-			s.finalizeRollappPacketsByReceiver(receiverAccount.String())
+			s.finalizeRollappPacketsByAddress(fulfillerAccount.String())
 			// Funds are passed to the fulfiller
 			fulfillerAccountBalanceAfterTimeout := bankKeeper.GetBalance(s.hubCtx(), fulfillerAccount, sdk.DefaultBondDenom)
 			s.Require().True(fulfillerAccountBalanceAfterTimeout.IsEqual(fulfillerInitialBalance.Add(lastDemandOrder.Fee[0])))

ibctesting/utils_test.go

@@ -373,16 +373,15 @@ func (s *utilSuite) newTestChainWithSingleValidator(t *testing.T, coord *ibctest
 	return chain
 }
 
-func (s *utilSuite) finalizeRollappPacketsByReceiver(receiver string) sdk.Events {
+func (s *utilSuite) finalizeRollappPacketsByAddress(address string) sdk.Events {
 	s.T().Helper()
-	// Query all pending packets by receiver
+	// Query all pending packets by address
 	querier := delayedackkeeper.NewQuerier(s.hubApp().DelayedAckKeeper)
-	resp, err := querier.GetPendingPacketsByReceiver(s.hubCtx(), &delayedacktypes.QueryPendingPacketsByReceiverRequest{
-		RollappId: rollappChainID(),
-		Receiver:  receiver,
+	resp, err := querier.GetPendingPacketsByAddress(s.hubCtx(), &delayedacktypes.QueryPendingPacketsByAddressRequest{
+		Address: address,
 	})
 	s.Require().NoError(err)
-	// Finalize all packets are collect events
+	// Finalize all packets and collect events
 	events := make(sdk.Events, 0)
 	for _, packet := range resp.RollappPackets {
 		k := common.EncodePacketKey(packet.RollappPacketKey())

proto/dymensionxyz/dymension/delayedack/events.proto

@@ -19,28 +19,3 @@ message EventFinalizePacket {
   // PacketSequence is a sequence number of the packet.
   uint64 packet_sequence = 6;
 }
-
-message EventFinalizePacketsUntilHeight {
-  // Sender is the signer of the message.
-  string sender = 1;
-  // RollappID is the ID of the rollapp.
-  string rollapp_id = 2;
-  // Height is a height until which packets are to be finalized. Height is inclusive.
-  uint64 height = 3;
-  // FinalizedNum is the number of finalized packets.
-  uint64 finalized_num = 4;
-}
-
-message EventFinalizeRollappPacketsByReceiver {
-  // Sender is the signer of the message.
-  string sender = 1;
-  // RollappID is the ID of the rollapp.
-  string rollapp_id = 2;
-  // Receiver is the one who waits tokens after the finalization.
-  string receiver = 3;
-  // Height is a height until which packets are to be finalized.
-  uint64 height = 4;
-  // FinalizedNum is the number of finalized packets.
-  uint64 finalized_num = 5;
-
-}

proto/dymensionxyz/dymension/delayedack/query.proto

@@ -23,8 +23,8 @@ service Query {
 	}
 
 	// Queries a list of pending RollappPacket items by rollappID and receiver.
-	rpc GetPendingPacketsByReceiver(QueryPendingPacketsByReceiverRequest) returns (QueryPendingPacketByReceiverListResponse) {
-		option (google.api.http).get = "/dymensionxyz/dymension/delayedack/pending-receiver-packets/{rollappId}/{receiver}";
+	rpc GetPendingPacketsByAddress(QueryPendingPacketsByAddressRequest) returns (QueryPendingPacketByAddressListResponse) {
+		option (google.api.http).get = "/dymensionxyz/dymension/delayedack/pending-receiver-packets/{address}";
 	}
 }
 
@@ -49,13 +49,12 @@ message QueryRollappPacketListResponse {
 	cosmos.base.query.v1beta1.PageResponse pagination = 2;
 }
 
-message QueryPendingPacketsByReceiverRequest {
-	string rollappId = 1;
-	string receiver = 2;
-	cosmos.base.query.v1beta1.PageRequest pagination = 3;
+message QueryPendingPacketsByAddressRequest {
+	string address = 1;
+	cosmos.base.query.v1beta1.PageRequest pagination = 2;
 }
 
-message QueryPendingPacketByReceiverListResponse {
+message QueryPendingPacketByAddressListResponse {
 	repeated common.RollappPacket rollappPackets = 1 [(gogoproto.nullable) = false];
 	cosmos.base.query.v1beta1.PageResponse pagination = 2;
 }

x/common/types/rollapp_packet.go

@@ -91,6 +91,14 @@ func (r RollappPacket) GetTransferPacketData() (transfertypes.FungibleTokenPacke
 	return data, nil
 }
 
+func (r RollappPacket) MustGetTransferPacketData() transfertypes.FungibleTokenPacketData {
+	data, err := r.GetTransferPacketData()
+	if err != nil {
+		panic(err)
+	}
+	return data
+}
+
 func (r RollappPacket) GetAck() (channeltypes.Acknowledgement, error) {
 	var ack channeltypes.Acknowledgement
 	if err := transfertypes.ModuleCdc.UnmarshalJSON(r.Acknowledgement, &ack); err != nil {

x/delayedack/client/cli/query.go

@@ -27,7 +27,7 @@ func GetQueryCmd() *cobra.Command {
 	cmd.AddCommand(CmdGetPacketsByRollapp())
 	cmd.AddCommand(CmdGetPacketsByStatus())
 	cmd.AddCommand(CmdGetPacketsByType())
-	cmd.AddCommand(CmdGetPendingPacketsByReceiver())
+	cmd.AddCommand(CmdGetPendingPacketsByAddress())
 
 	return cmd
 }
@@ -219,21 +219,20 @@ func CmdGetPacketsByType() *cobra.Command {
 	return cmd
 }
 
-func CmdGetPendingPacketsByReceiver() *cobra.Command {
+func CmdGetPendingPacketsByAddress() *cobra.Command {
 	cmd := &cobra.Command{
-		Use:   "pending-packets-by-receiver [rollapp-id] [receiver]",
-		Short: "Get pending packets by receiver",
-		Args:  cobra.MinimumNArgs(2),
+		Use:   "pending-packets-by-address [address]",
+		Short: "Get pending packets by address",
+		Args:  cobra.MinimumNArgs(1),
 		RunE: func(cmd *cobra.Command, args []string) error {
 			clientCtx, err := client.GetClientQueryContext(cmd)
 			if err != nil {
 				return err
 			}
 			queryClient := types.NewQueryClient(clientCtx)
 
-			res, err := queryClient.GetPendingPacketsByReceiver(cmd.Context(), &types.QueryPendingPacketsByReceiverRequest{
-				RollappId:  args[0],
-				Receiver:   args[1],
+			res, err := queryClient.GetPendingPacketsByAddress(cmd.Context(), &types.QueryPendingPacketsByAddressRequest{
+				Address:    args[0],
 				Pagination: nil, // TODO: handle pagination
 			})
 			if err != nil {

x/delayedack/genesis.go

@@ -2,6 +2,8 @@ package delayedack
 
 import (
 	sdk "github.com/cosmos/cosmos-sdk/types"
+
+	commontypes "github.com/dymensionxyz/dymension/v3/x/common/types"
 	"github.com/dymensionxyz/dymension/v3/x/delayedack/keeper"
 	"github.com/dymensionxyz/dymension/v3/x/delayedack/types"
 )
@@ -10,6 +12,15 @@ import (
 func InitGenesis(ctx sdk.Context, k keeper.Keeper, genState types.GenesisState) {
 	k.SetParams(ctx, genState.Params)
 	for _, packet := range genState.RollappPackets {
+		transferPacketData := packet.MustGetTransferPacketData()
+		switch packet.Type {
+		case commontypes.RollappPacket_ON_RECV:
+			k.MustSetPendingPacketByAddress(ctx, transferPacketData.Receiver, packet.RollappPacketKey())
+		case commontypes.RollappPacket_ON_ACK, commontypes.RollappPacket_ON_TIMEOUT:
+			k.MustSetPendingPacketByAddress(ctx, transferPacketData.Sender, packet.RollappPacketKey())
+		case commontypes.RollappPacket_UNDEFINED:
+			panic("invalid rollapp packet type")
+		}
 		k.SetRollappPacket(ctx, packet)
 	}
 }