feat!: optimize pending packets storage on consumer + migration

proto/interchain_security/ccv/v1/ccv.proto

@@ -58,17 +58,19 @@ message MaturedUnbondingOps {
   repeated uint64 ids = 1;
 }
 
-// ConsumerPacketData contains a consumer packet data and a type tag
+// ConsumerPacketData contains a consumer packet data, type tag, and index for storage.
 message ConsumerPacketData {
   ConsumerPacketDataType type = 1; 
 
   oneof data {
     SlashPacketData slashPacketData = 2;
     VSCMaturedPacketData vscMaturedPacketData = 3;
   }
+  uint64 idx = 4;
 }
 
 
+// [Depreciated] favor using []ConsumerPacketData directly.
 // ConsumerPacketDataList is a list of consumer packet data packets.
 message ConsumerPacketDataList {
   repeated ConsumerPacketData list = 1

tests/integration/expired_client.go

@@ -122,7 +122,7 @@ func (s *CCVTestSuite) TestConsumerPacketSendExpiredClient() {
 	// check that the packets were added to the list of pending data packets
 	consumerPackets := consumerKeeper.GetPendingPackets(s.consumerCtx())
 	s.Require().NotEmpty(consumerPackets)
-	s.Require().Equal(2, len(consumerPackets.GetList()), "unexpected number of pending data packets")
+	s.Require().Len(consumerPackets, 2, "unexpected number of pending data packets")
 
 	// try to send slash packet for downtime infraction
 	addr := ed25519.GenPrivKey().PubKey().Address()
@@ -136,7 +136,7 @@ func (s *CCVTestSuite) TestConsumerPacketSendExpiredClient() {
 	// check that the packets were added to the list of pending data packets
 	consumerPackets = consumerKeeper.GetPendingPackets(s.consumerCtx())
 	s.Require().NotEmpty(consumerPackets)
-	s.Require().Equal(4, len(consumerPackets.GetList()), "unexpected number of pending data packets")
+	s.Require().Len(consumerPackets, 4, "unexpected number of pending data packets")
 
 	// upgrade expired client to the consumer
 	upgradeExpiredClient(s, Provider)
@@ -147,7 +147,6 @@ func (s *CCVTestSuite) TestConsumerPacketSendExpiredClient() {
 	// check that the list of pending data packets is emptied
 	consumerPackets = consumerKeeper.GetPendingPackets(s.consumerCtx())
 	s.Require().Empty(consumerPackets)
-	s.Require().Equal(0, len(consumerPackets.GetList()), "unexpected number of pending data packets")
 
 	// relay all  packet from consumer to provider
 	relayAllCommittedPackets(s, s.consumerChain, s.path, ccv.ConsumerPortID, s.path.EndpointA.ChannelID, 4)

tests/integration/slashing.go

@@ -486,15 +486,15 @@ func (suite *CCVTestSuite) TestValidatorDowntime() {
 
 	// check that slash packet is queued
 	pendingPackets := consumerKeeper.GetPendingPackets(ctx)
-	suite.Require().NotEmpty(pendingPackets.List, "pending packets empty")
-	suite.Require().Len(pendingPackets.List, 1, "pending packets len should be 1 is %d", len(pendingPackets.List))
+	suite.Require().NotEmpty(pendingPackets, "pending packets empty")
+	suite.Require().Len(pendingPackets, 1, "pending packets len should be 1 is %d", len(pendingPackets))
 
 	// clear queue, commit packets
 	suite.consumerApp.GetConsumerKeeper().SendPackets(ctx)
 
 	// check queue was cleared
 	pendingPackets = suite.consumerApp.GetConsumerKeeper().GetPendingPackets(ctx)
-	suite.Require().Empty(pendingPackets.List, "pending packets NOT empty")
+	suite.Require().Empty(pendingPackets, "pending packets NOT empty")
 
 	// verify that the slash packet was sent
 	gotCommit := consumerIBCKeeper.ChannelKeeper.GetPacketCommitment(ctx, ccv.ConsumerPortID, channelID, seq)
@@ -573,15 +573,15 @@ func (suite *CCVTestSuite) TestValidatorDoubleSigning() {
 
 	// check slash packet is queued
 	pendingPackets := suite.consumerApp.GetConsumerKeeper().GetPendingPackets(ctx)
-	suite.Require().NotEmpty(pendingPackets.List, "pending packets empty")
-	suite.Require().Len(pendingPackets.List, 1, "pending packets len should be 1 is %d", len(pendingPackets.List))
+	suite.Require().NotEmpty(pendingPackets, "pending packets empty")
+	suite.Require().Len(pendingPackets, 1, "pending packets len should be 1 is %d", len(pendingPackets))
 
 	// clear queue, commit packets
 	suite.consumerApp.GetConsumerKeeper().SendPackets(ctx)
 
 	// check queue was cleared
 	pendingPackets = suite.consumerApp.GetConsumerKeeper().GetPendingPackets(ctx)
-	suite.Require().Empty(pendingPackets.List, "pending packets NOT empty")
+	suite.Require().Empty(pendingPackets, "pending packets NOT empty")
 
 	// check slash packet is sent
 	gotCommit := suite.consumerApp.GetIBCKeeper().ChannelKeeper.GetPacketCommitment(ctx, ccv.ConsumerPortID, channelID, seq)
@@ -636,7 +636,7 @@ func (suite *CCVTestSuite) TestQueueAndSendSlashPacket() {
 	// the downtime slash request duplicates
 	dataPackets := consumerKeeper.GetPendingPackets(ctx)
 	suite.Require().NotEmpty(dataPackets)
-	suite.Require().Len(dataPackets.GetList(), 12)
+	suite.Require().Len(dataPackets, 12)
 
 	// save consumer next sequence
 	seq, _ := consumerIBCKeeper.ChannelKeeper.GetNextSequenceSend(ctx, ccv.ConsumerPortID, channelID)
@@ -663,7 +663,7 @@ func (suite *CCVTestSuite) TestQueueAndSendSlashPacket() {
 	// check that pending data packets got cleared
 	dataPackets = consumerKeeper.GetPendingPackets(ctx)
 	suite.Require().Empty(dataPackets)
-	suite.Require().Len(dataPackets.GetList(), 0)
+	suite.Require().Len(dataPackets, 0)
 }
 
 // TestCISBeforeCCVEstablished tests that the consumer chain doesn't panic or
@@ -674,14 +674,14 @@ func (suite *CCVTestSuite) TestCISBeforeCCVEstablished() {
 
 	// Check pending packets is empty
 	pendingPackets := consumerKeeper.GetPendingPackets(suite.consumerCtx())
-	suite.Require().Len(pendingPackets.List, 0)
+	suite.Require().Len(pendingPackets, 0)
 
 	consumerKeeper.Slash(suite.consumerCtx(), []byte{0x01, 0x02, 0x3},
 		66, 4324, sdk.MustNewDecFromStr("0.05"), stakingtypes.Downtime)
 
 	// Check slash packet was queued
 	pendingPackets = consumerKeeper.GetPendingPackets(suite.consumerCtx())
-	suite.Require().Len(pendingPackets.List, 1)
+	suite.Require().Len(pendingPackets, 1)
 
 	// Pass 5 blocks, confirming the consumer doesn't panic
 	for i := 0; i < 5; i++ {
@@ -690,7 +690,7 @@ func (suite *CCVTestSuite) TestCISBeforeCCVEstablished() {
 
 	// Check packet is still queued
 	pendingPackets = consumerKeeper.GetPendingPackets(suite.consumerCtx())
-	suite.Require().Len(pendingPackets.List, 1)
+	suite.Require().Len(pendingPackets, 1)
 
 	// establish ccv channel
 	suite.SetupCCVChannel(suite.path)
@@ -699,5 +699,5 @@ func (suite *CCVTestSuite) TestCISBeforeCCVEstablished() {
 	// Pass one more block, and confirm the packet is sent now that ccv channel is established
 	suite.consumerChain.NextBlock()
 	pendingPackets = consumerKeeper.GetPendingPackets(suite.consumerCtx())
-	suite.Require().Len(pendingPackets.List, 0)
+	suite.Require().Len(pendingPackets, 0)
 }

x/ccv/consumer/keeper/genesis.go

@@ -89,9 +89,12 @@ func (k Keeper) InitGenesis(ctx sdk.Context, state *consumertypes.GenesisState)
 			k.SetLastTransmissionBlockHeight(ctx, state.LastTransmissionBlockHeight)
 		}
 
-		// set pending consumer pending packets
+		// Set pending consumer packets, using the depreciated ConsumerPacketDataList type
+		// that exists for genesis.
 		// note that the list includes pending mature VSC packet only if the handshake is completed
-		k.AppendPendingPacket(ctx, state.PendingConsumerPackets.List...)
+		for _, packet := range state.PendingConsumerPackets.List {
+			k.AppendPendingPacket(ctx, packet.Type, packet.Data)
+		}
 
 		// set height to valset update id mapping
 		for _, h2v := range state.HeightToValsetUpdateId {
@@ -122,6 +125,11 @@ func (k Keeper) ExportGenesis(ctx sdk.Context) (genesis *consumertypes.GenesisSt
 	// export the current validator set
 	valset := k.MustGetCurrentValidatorsAsABCIUpdates(ctx)
 
+	// export pending packets using the depreciated ConsumerPacketDataList type
+	pendingPackets := k.GetPendingPackets(ctx)
+	pendingPacketsDepreciated := ccv.ConsumerPacketDataList{}
+	pendingPacketsDepreciated.List = append(pendingPacketsDepreciated.List, pendingPackets...)
+
 	// export all the states created after a provider channel got established
 	if channelID, ok := k.GetProviderChannel(ctx); ok {
 		clientID, found := k.GetProviderClientID(ctx)
@@ -136,7 +144,7 @@ func (k Keeper) ExportGenesis(ctx sdk.Context) (genesis *consumertypes.GenesisSt
 			k.GetAllPacketMaturityTimes(ctx),
 			valset,
 			k.GetAllHeightToValsetUpdateIDs(ctx),
-			k.GetPendingPackets(ctx),
+			pendingPacketsDepreciated,
 			k.GetAllOutstandingDowntimes(ctx),
 			k.GetLastTransmissionBlockHeight(ctx),
 			params,
@@ -156,7 +164,7 @@ func (k Keeper) ExportGenesis(ctx sdk.Context) (genesis *consumertypes.GenesisSt
 			nil,
 			valset,
 			k.GetAllHeightToValsetUpdateIDs(ctx),
-			k.GetPendingPackets(ctx),
+			pendingPacketsDepreciated,
 			nil,
 			consumertypes.LastTransmissionBlockHeight{},
 			params,

x/ccv/consumer/keeper/genesis_test.go

@@ -148,7 +148,12 @@ func TestInitGenesis(t *testing.T) {
 			func(ctx sdk.Context, ck consumerkeeper.Keeper, gs *consumertypes.GenesisState) {
 				assertConsumerPortIsBound(t, ctx, &ck)
 
-				require.Equal(t, pendingDataPackets, ck.GetPendingPackets(ctx))
+				obtainedPendingPackets := ck.GetPendingPackets(ctx)
+				for idx, expectedPacketData := range pendingDataPackets.List {
+					require.Equal(t, expectedPacketData.Type, obtainedPendingPackets[idx].Type)
+					require.Equal(t, expectedPacketData.Data, obtainedPendingPackets[idx].Data)
+				}
+
 				assertHeightValsetUpdateIDs(t, ctx, &ck, defaultHeightValsetUpdateIDs)
 				assertProviderClientID(t, ctx, &ck, provClientID)
 				require.Equal(t, validator.Address.Bytes(), ck.GetAllCCValidator(ctx)[0].Address)
@@ -186,7 +191,12 @@ func TestInitGenesis(t *testing.T) {
 				require.Equal(t, provChannelID, gotChannelID)
 
 				require.True(t, ck.PacketMaturityTimeExists(ctx, matPackets[0].VscId, matPackets[0].MaturityTime))
-				require.Equal(t, pendingDataPackets, ck.GetPendingPackets(ctx))
+
+				obtainedPendingPackets := ck.GetPendingPackets(ctx)
+				for idx, expectedPacketData := range pendingDataPackets.List {
+					require.Equal(t, expectedPacketData.Type, obtainedPendingPackets[idx].Type)
+					require.Equal(t, expectedPacketData.Data, obtainedPendingPackets[idx].Data)
+				}
 
 				require.Equal(t, gs.OutstandingDowntimeSlashing, ck.GetAllOutstandingDowntimes(ctx))
 
@@ -252,12 +262,16 @@ func TestExportGenesis(t *testing.T) {
 				Data: &ccv.ConsumerPacketData_SlashPacketData{
 					SlashPacketData: ccv.NewSlashPacketData(abciValidator, vscID, stakingtypes.Downtime),
 				},
+				Idx: 0,
 			},
 			{
 				Type: ccv.VscMaturedPacket,
 				Data: &ccv.ConsumerPacketData_VscMaturedPacketData{
 					VscMaturedPacketData: ccv.NewVSCMaturedPacketData(vscID),
 				},
+				// This idx is a part of the expected genesis state.
+				// If the keeper is correctly storing consumer packet data, indexes should be populated.
+				Idx: 1,
 			},
 		},
 	}
@@ -291,7 +305,10 @@ func TestExportGenesis(t *testing.T) {
 				ck.SetCCValidator(ctx, cVal)
 				ck.SetParams(ctx, params)
 
-				ck.AppendPendingPacket(ctx, consPackets.List...)
+				for _, packet := range consPackets.List {
+					ck.AppendPendingPacket(ctx, packet.Type, packet.Data)
+				}
+
 				ck.SetHeightValsetUpdateID(ctx, defaultHeightValsetUpdateIDs[0].Height, defaultHeightValsetUpdateIDs[0].ValsetUpdateId)
 			},
 			consumertypes.NewRestartGenesisState(
@@ -321,7 +338,9 @@ func TestExportGenesis(t *testing.T) {
 				ck.SetHeightValsetUpdateID(ctx, updatedHeightValsetUpdateIDs[0].Height, updatedHeightValsetUpdateIDs[0].ValsetUpdateId)
 				ck.SetHeightValsetUpdateID(ctx, updatedHeightValsetUpdateIDs[1].Height, updatedHeightValsetUpdateIDs[1].ValsetUpdateId)
 
-				ck.AppendPendingPacket(ctx, consPackets.List...)
+				for _, packet := range consPackets.List {
+					ck.AppendPendingPacket(ctx, packet.Type, packet.Data)
+				}
 
 				// populate the required states for an established CCV channel
 				ck.SetPacketMaturityTime(ctx, matPackets[0].VscId, matPackets[0].MaturityTime)

x/ccv/consumer/keeper/keeper.go

@@ -565,48 +565,78 @@ func (k Keeper) GetAllCCValidator(ctx sdk.Context) (validators []types.CrossChai
 	return validators
 }
 
-// SetPendingPackets sets the pending CCV packets
-func (k Keeper) SetPendingPackets(ctx sdk.Context, packets ccv.ConsumerPacketDataList) {
-	store := ctx.KVStore(k.storeKey)
-	bz, err := packets.Marshal()
-	if err != nil {
-		// This should never happen
-		panic(fmt.Errorf("failed to marshal ConsumerPacketDataList: %w", err))
-	}
-	store.Set(types.PendingDataPacketsKey(), bz)
+// Note: PendingDataPacketsBytePrefix is the correct prefix, NOT PendingDataPacketsByteKey.
+// See consistency with PendingDataPacketsKey().
+func PendingDataPacketsIterator(store sdk.KVStore) sdk.Iterator {
+	return sdk.KVStorePrefixIterator(store, []byte{types.PendingDataPacketsBytePrefix})
 }
 
-// GetPendingPackets returns the pending CCV packets from the store
-func (k Keeper) GetPendingPackets(ctx sdk.Context) ccv.ConsumerPacketDataList {
-	var packets ccv.ConsumerPacketDataList
-
+func (k Keeper) getAndIncrementPendingPacketsIdx(ctx sdk.Context) (toReturn uint64) {
 	store := ctx.KVStore(k.storeKey)
-	bz := store.Get(types.PendingDataPacketsKey())
+	bz := store.Get(types.PendingPacketsIndexKey())
 	if bz == nil {
-		return packets
+		toReturn = 0
+	} else {
+		toReturn = binary.BigEndian.Uint64(bz)
 	}
+	toStore := toReturn + 1
+	store.Set(types.PendingPacketsIndexKey(), sdk.Uint64ToBigEndian(toStore))
+	return toReturn
+}
 
-	err := packets.Unmarshal(bz)
-	if err != nil {
-		// An error here would indicate something is very wrong,
-		// the PendingPackets are assumed to be correctly serialized in SetPendingPackets.
-		panic(fmt.Errorf("failed to unmarshal pending data packets: %w", err))
+// GetPendingPackets returns ALL the pending CCV packets from the store
+func (k Keeper) GetPendingPackets(ctx sdk.Context) []ccv.ConsumerPacketData {
+	var packets []ccv.ConsumerPacketData
+	store := ctx.KVStore(k.storeKey)
+	iterator := PendingDataPacketsIterator(store)
+	defer iterator.Close()
+	for ; iterator.Valid(); iterator.Next() {
+		var packet ccv.ConsumerPacketData
+		bz := iterator.Value()
+		err := packet.Unmarshal(bz)
+		if err != nil {
+			// An error here would indicate something is very wrong,
+			panic(fmt.Errorf("failed to unmarshal pending data packet: %w", err))
+		}
+		packets = append(packets, packet)
 	}
-
 	return packets
 }
 
-// DeletePendingDataPackets clears the pending data packets in store
-func (k Keeper) DeletePendingDataPackets(ctx sdk.Context) {
+// DeletePendingDataPackets deletes pending data packets with given indexes
+func (k Keeper) DeletePendingDataPackets(ctx sdk.Context, idxs ...uint64) {
+	store := ctx.KVStore(k.storeKey)
+	for _, idx := range idxs {
+		store.Delete(types.PendingDataPacketsKey(idx))
+	}
+}
+
+func (k Keeper) DeleteAllPendingDataPackets(ctx sdk.Context) {
 	store := ctx.KVStore(k.storeKey)
-	store.Delete(types.PendingDataPacketsKey())
+	// Note: PendingDataPacketsBytePrefix is the correct prefix, NOT PendingDataPacketsByteKey.
+	// See consistency with PendingDataPacketsKey().
+	iterator := PendingDataPacketsIterator(store)
+	defer iterator.Close()
+	for ; iterator.Valid(); iterator.Next() {
+		store.Delete(iterator.Key())
+	}
 }
 
-// AppendPendingDataPacket appends the given data packet to the pending data packets in store
-func (k Keeper) AppendPendingPacket(ctx sdk.Context, packet ...ccv.ConsumerPacketData) {
-	pending := k.GetPendingPackets(ctx)
-	list := append(pending.GetList(), packet...)
-	k.SetPendingPackets(ctx, ccv.ConsumerPacketDataList{List: list})
+// AppendPendingPacket enqueues the given data packet to the end of the pending data packets queue
+func (k Keeper) AppendPendingPacket(ctx sdk.Context, packetType ccv.ConsumerPacketDataType, data ccv.ExportedIsConsumerPacketData_Data) {
+	cpd := ccv.NewConsumerPacketData(
+		packetType,
+		data,
+		k.getAndIncrementPendingPacketsIdx(ctx),
+	)
+	key := types.PendingDataPacketsKey(cpd.Idx)
+	store := ctx.KVStore(k.storeKey)
+	bz, err := cpd.Marshal()
+	if err != nil {
+		// This should never happen
+		panic(fmt.Errorf("failed to marshal ConsumerPacketData: %w", err))
+	}
+	store.Set(key, bz)
 }
 
 func (k Keeper) MarkAsPrevStandaloneChain(ctx sdk.Context) {