Throttle garbage collection (#557)

* changes

* Update proposal.go

* add log

* Update throttle_test.go

* fixes

tests/e2e/common.go

@@ -33,7 +33,11 @@ const (
 
 // firstConsumerBundle returns the bundle of the first consumer chain
 func (s *CCVTestSuite) getFirstBundle() icstestingutils.ConsumerBundle {
-	return *s.consumerBundles[icstestingutils.FirstConsumerChainID]
+	return s.getBundleByIdx(0)
+}
+
+func (s *CCVTestSuite) getBundleByIdx(index int) icstestingutils.ConsumerBundle {
+	return *s.consumerBundles[ibctesting.GetChainID(2+index)]
 }
 
 func (s *CCVTestSuite) providerCtx() sdk.Context {

tests/e2e/stop_consumer.go

@@ -4,6 +4,7 @@ import (
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	stakingtypes "github.com/cosmos/cosmos-sdk/x/staking/types"
 	channeltypes "github.com/cosmos/ibc-go/v3/modules/core/04-channel/types"
+	"github.com/cosmos/interchain-security/x/ccv/provider/types"
 	ccv "github.com/cosmos/interchain-security/x/ccv/types"
 	abci "github.com/tendermint/tendermint/abci/types"
 )
@@ -14,8 +15,7 @@ func (s *CCVTestSuite) TestStopConsumerChain() {
 	providerKeeper := s.providerApp.GetProviderKeeper()
 	providerStakingKeeper := s.providerApp.GetE2eStakingKeeper()
 
-	// default consumer chain ID
-	consumerChainID := s.consumerChain.ChainID
+	firstBundle := s.getFirstBundle()
 
 	// choose a validator
 	tmValidator := s.providerChain.Vals.Validators[0]
@@ -48,7 +48,7 @@ func (s *CCVTestSuite) TestStopConsumerChain() {
 	}{
 		{
 			func(suite *CCVTestSuite) error {
-				suite.SetupCCVChannel(s.path)
+				suite.SetupAllCCVChannels()
 				suite.SetupTransferChannel()
 				return nil
 			},
@@ -75,9 +75,33 @@ func (s *CCVTestSuite) TestStopConsumerChain() {
 		},
 		{
 			func(suite *CCVTestSuite) error {
-				providerKeeper.SetSlashAcks(s.providerCtx(), consumerChainID, []string{"validator-1", "validator-2", "validator-3"})
-				providerKeeper.SetLockUnbondingOnTimeout(s.providerCtx(), consumerChainID)
-				providerKeeper.AppendPendingPackets(s.providerCtx(), consumerChainID, ccv.ValidatorSetChangePacketData{ValsetUpdateId: 1})
+				providerKeeper.SetSlashAcks(s.providerCtx(), firstBundle.Chain.ChainID, []string{"validator-1", "validator-2", "validator-3"})
+				providerKeeper.SetLockUnbondingOnTimeout(s.providerCtx(), firstBundle.Chain.ChainID)
+				providerKeeper.AppendPendingPackets(s.providerCtx(), firstBundle.Chain.ChainID, ccv.ValidatorSetChangePacketData{ValsetUpdateId: 1})
+				return nil
+			},
+		},
+		{
+			func(suite *CCVTestSuite) error {
+
+				// Queue slash and vsc packet data for consumer 0, these queue entries will be removed
+				firstBundle := s.getFirstBundle()
+				globalEntry := types.NewSlashPacketEntry(s.providerCtx().BlockTime(), firstBundle.Chain.ChainID, 7, []byte{})
+				providerKeeper.QueuePendingSlashPacketEntry(s.providerCtx(), globalEntry)
+				providerKeeper.QueuePendingSlashPacketData(s.providerCtx(), firstBundle.Chain.ChainID, 1,
+					ccv.SlashPacketData{ValsetUpdateId: 1})
+				providerKeeper.QueuePendingVSCMaturedPacketData(s.providerCtx(),
+					firstBundle.Chain.ChainID, 2, ccv.VSCMaturedPacketData{ValsetUpdateId: 2})
+
+				// Queue slash and vsc packet data for consumer 1, these queue entries will be not be removed
+				secondBundle := s.getBundleByIdx(1)
+				globalEntry = types.NewSlashPacketEntry(s.providerCtx().BlockTime(), secondBundle.Chain.ChainID, 7, []byte{})
+				providerKeeper.QueuePendingSlashPacketEntry(s.providerCtx(), globalEntry)
+				providerKeeper.QueuePendingSlashPacketData(s.providerCtx(), secondBundle.Chain.ChainID, 1,
+					ccv.SlashPacketData{ValsetUpdateId: 1})
+				providerKeeper.QueuePendingVSCMaturedPacketData(s.providerCtx(),
+					secondBundle.Chain.ChainID, 2, ccv.VSCMaturedPacketData{ValsetUpdateId: 2})
+
 				return nil
 			},
 		},
@@ -89,11 +113,20 @@ func (s *CCVTestSuite) TestStopConsumerChain() {
 	}
 
 	// stop the consumer chain
-	err = providerKeeper.StopConsumerChain(s.providerCtx(), consumerChainID, false, true)
+	err = providerKeeper.StopConsumerChain(s.providerCtx(), firstBundle.Chain.ChainID, false, true)
 	s.Require().NoError(err)
 
 	// check all states are removed and the unbonding operation released
-	s.checkConsumerChainIsRemoved(consumerChainID, false, true)
+	s.checkConsumerChainIsRemoved(firstBundle.Chain.ChainID, false, true)
+
+	// check entries related to second consumer chain are not removed
+	s.Require().Len(providerKeeper.GetAllPendingSlashPacketEntries(s.providerCtx()), 1)
+
+	secondBundle := s.getBundleByIdx(1)
+	slashData, vscMaturedData := providerKeeper.GetAllPendingPacketData(
+		s.providerCtx(), secondBundle.Chain.ChainID)
+	s.Require().Len(slashData, 1)
+	s.Require().Len(vscMaturedData, 1)
 }
 
 // TODO Simon: implement OnChanCloseConfirm in IBC-GO testing to close the consumer chain's channel end
@@ -173,6 +206,17 @@ func (s *CCVTestSuite) checkConsumerChainIsRemoved(chainID string, lockUbd bool,
 	s.Require().Nil(providerKeeper.GetSlashAcks(s.providerCtx(), chainID))
 	s.Require().Zero(providerKeeper.GetInitChainHeight(s.providerCtx(), chainID))
 	s.Require().Nil(providerKeeper.GetPendingPackets(s.providerCtx(), chainID))
+
+	// No remaining global entries for this consumer
+	allGlobalEntries := providerKeeper.GetAllPendingSlashPacketEntries(s.providerCtx())
+	for _, entry := range allGlobalEntries {
+		s.Require().NotEqual(chainID, entry.ConsumerChainID)
+	}
+
+	// No remaining per-chain entries for this consumer
+	slashData, vscMaturedData := providerKeeper.GetAllPendingPacketData(s.providerCtx(), chainID)
+	s.Require().Empty(slashData)
+	s.Require().Empty(vscMaturedData)
 }
 
 // TestProviderChannelClosed checks that a consumer chain panics

x/ccv/provider/keeper/proposal.go

@@ -218,6 +218,20 @@ func (k Keeper) StopConsumerChain(ctx sdk.Context, chainID string, lockUbd, clos
 		k.DeleteUnbondingOpIndex(ctx, chainID, id)
 	}
 
+	// Remove any existing throttling related queue entries from the
+	// global queue for this consumer.
+	k.DeletePendingSlashPacketEntriesForConsumer(ctx, chainID)
+
+	if k.GetPendingPacketDataSize(ctx, chainID) > 0 {
+		k.Logger(ctx).Info("There are pending slash packets queued,"+
+			" from a consumer that is being removed. Those slash packets will be thrown out!", "chainID", chainID)
+	}
+
+	// Remove all pending slash packets and vsc matured packets queued for this consumer.
+	// Note: queued VSC matured packets can be safely removed from the per-chain queue,
+	// since all unbonding operations for this consumer are release above.
+	k.DeletePendingPacketDataForConsumer(ctx, chainID)
+
 	return nil
 }
 

x/ccv/provider/keeper/proposal_test.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	_go "github.com/confio/ics23/go"
+	"github.com/cosmos/cosmos-sdk/crypto/keys/ed25519"
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	clienttypes "github.com/cosmos/ibc-go/v3/modules/core/02-client/types"
 	ibctmtypes "github.com/cosmos/ibc-go/v3/modules/light-clients/07-tendermint/types"
@@ -420,6 +421,19 @@ func TestStopConsumerChain(t *testing.T) {
 			},
 			expErr: false,
 		},
+		{
+			description: "valid stop of consumer chain, throttle related queues are cleaned",
+			setup: func(ctx sdk.Context, providerKeeper *providerkeeper.Keeper, mocks testkeeper.MockedKeepers) {
+
+				testkeeper.SetupForStoppingConsumerChain(t, ctx, providerKeeper, mocks)
+
+				providerKeeper.QueuePendingSlashPacketEntry(ctx, providertypes.NewSlashPacketEntry(
+					ctx.BlockTime(), "chainID", 1, ed25519.GenPrivKey().PubKey().Address()))
+				providerKeeper.QueuePendingSlashPacketData(ctx, "chainID", 1, testkeeper.GetNewSlashPacketData())
+				providerKeeper.QueuePendingVSCMaturedPacketData(ctx, "chainID", 2, testkeeper.GetNewVSCMaturedPacketData())
+			},
+			expErr: false,
+		},
 		{
 			description: "valid stop of consumer chain, all mock calls hit",
 			setup: func(ctx sdk.Context, providerKeeper *providerkeeper.Keeper, mocks testkeeper.MockedKeepers) {
@@ -503,6 +517,15 @@ func testProviderStateIsCleaned(t *testing.T, ctx sdk.Context, providerKeeper pr
 		return true // stop the iteration
 	})
 	require.False(t, found)
+
+	allGlobalEntries := providerKeeper.GetAllPendingSlashPacketEntries(ctx)
+	for _, entry := range allGlobalEntries {
+		require.NotEqual(t, expectedChainID, entry.ConsumerChainID)
+	}
+
+	slashPacketData, vscMaturedPacketData := providerKeeper.GetAllPendingPacketData(ctx, expectedChainID)
+	require.Empty(t, slashPacketData)
+	require.Empty(t, vscMaturedPacketData)
 }
 
 // TestPendingConsumerRemovalPropDeletion tests the getting/setting

x/ccv/provider/keeper/throttle.go

@@ -213,6 +213,22 @@ func (k Keeper) GetAllPendingSlashPacketEntries(ctx sdktypes.Context) (entries [
 	return entries
 }
 
+// DeleteAllPendingSlashPacketEntries deletes all pending slash packet entries in the global queue,
+// only relevant to a single consumer.
+func (k Keeper) DeletePendingSlashPacketEntriesForConsumer(ctx sdktypes.Context, consumerChainID string) {
+	entriesToDel := []providertypes.SlashPacketEntry{}
+	k.IteratePendingSlashPacketEntries(ctx, func(entry providertypes.SlashPacketEntry) (stop bool) {
+		if entry.ConsumerChainID == consumerChainID {
+			entriesToDel = append(entriesToDel, entry)
+		}
+		// Continue iteration
+		stop = false
+		return stop
+	})
+
+	k.DeletePendingSlashPacketEntries(ctx, entriesToDel...)
+}
+
 // IteratePendingSlashPackets iterates over the pending slash packet entry queue and calls the provided callback
 func (k Keeper) IteratePendingSlashPacketEntries(ctx sdktypes.Context,
 	cb func(providertypes.SlashPacketEntry) (stop bool)) {
@@ -366,6 +382,44 @@ func (k Keeper) IteratePendingPacketData(ctx sdktypes.Context, consumerChainID s
 	}
 }
 
+// GetAllPendingPacketData returns all pending packet data for a specific consumer chain.
+//
+// Note: This method is only used by tests
+func (k Keeper) GetAllPendingPacketData(ctx sdktypes.Context, consumerChainID string) (
+	[]ccvtypes.SlashPacketData, []ccvtypes.VSCMaturedPacketData) {
+
+	slashData := []ccvtypes.SlashPacketData{}
+	vscMaturedData := []ccvtypes.VSCMaturedPacketData{}
+	k.IteratePendingPacketData(ctx, consumerChainID, func(ibcSeqNum uint64, data interface{}) (stop bool) {
+
+		switch data := data.(type) {
+
+		case ccvtypes.SlashPacketData:
+			slashData = append(slashData, data)
+		case ccvtypes.VSCMaturedPacketData:
+			vscMaturedData = append(vscMaturedData, data)
+		default:
+			panic(fmt.Sprintf("unexpected pending packet data type: %T", data))
+		}
+		// Continue iteration
+		stop = false
+		return stop
+	})
+	return slashData, vscMaturedData
+}
+
+// DeletePendingPacketDataForConsumer deletes all pending packet data for the given consumer chain.
+func (k Keeper) DeletePendingPacketDataForConsumer(ctx sdktypes.Context, consumerChainID string) {
+	ibcSeqNumsToDelete := []uint64{}
+	k.IteratePendingPacketData(ctx, consumerChainID, func(ibcSeqNum uint64, packetData interface{}) bool {
+		ibcSeqNumsToDelete = append(ibcSeqNumsToDelete, ibcSeqNum)
+		// Continue iteration
+		stop := false
+		return stop
+	})
+	k.DeletePendingPacketData(ctx, consumerChainID, ibcSeqNumsToDelete...)
+}
+
 // DeletePendingPacketData deletes the given entries (specified by their ibc seq number) from the pending packet data queue
 func (k Keeper) DeletePendingPacketData(ctx sdktypes.Context, consumerChainID string, ibcSeqNumbers ...uint64) {
 	store := ctx.KVStore(k.storeKey)

x/ccv/provider/keeper/throttle_test.go

@@ -716,6 +716,37 @@ func TestPendingSlashPacketEntries(t *testing.T) {
 	require.Equal(t, 7, len(entries))
 }
 
+// Tests DeletePendingSlashPacketEntriesForConsumer.
+func TestDeletePendingSlashPacketEntriesForConsumer(t *testing.T) {
+
+	providerKeeper, ctx, ctrl, _ := testkeeper.GetProviderKeeperAndCtx(
+		t, testkeeper.NewInMemKeeperParams(t))
+	defer ctrl.Finish()
+
+	// Queue 2 global entries for a consumer chain ID
+	providerKeeper.QueuePendingSlashPacketEntry(ctx,
+		providertypes.NewSlashPacketEntry(time.Now().Add(time.Hour), "chain-78", 1,
+			ed25519.GenPrivKey().PubKey().Address()))
+	providerKeeper.QueuePendingSlashPacketEntry(ctx,
+		providertypes.NewSlashPacketEntry(time.Now().Add(time.Hour), "chain-78", 2,
+			ed25519.GenPrivKey().PubKey().Address()))
+
+	// Queue 1 global entry for two other consumer chain IDs
+	providerKeeper.QueuePendingSlashPacketEntry(ctx,
+		providertypes.NewSlashPacketEntry(time.Now().Add(2*time.Hour), "chain-79", 1,
+			ed25519.GenPrivKey().PubKey().Address()))
+	providerKeeper.QueuePendingSlashPacketEntry(ctx,
+		providertypes.NewSlashPacketEntry(time.Now().Add(3*time.Hour), "chain-80", 1,
+			ed25519.GenPrivKey().PubKey().Address()))
+
+	// Delete entries for chain-78, confirm those are deleted, and the other two remain
+	providerKeeper.DeletePendingSlashPacketEntriesForConsumer(ctx, "chain-78")
+	allEntries := providerKeeper.GetAllPendingSlashPacketEntries(ctx)
+	require.Equal(t, 2, len(allEntries))
+	require.Equal(t, "chain-79", allEntries[0].ConsumerChainID)
+	require.Equal(t, "chain-80", allEntries[1].ConsumerChainID)
+}
+
 // TestPendingSlashPacketEntryDeletion tests the deletion function for
 // pending slash packet entries with assertion of FIFO ordering.
 func TestPendingSlashPacketEntryDeletion(t *testing.T) {
@@ -869,6 +900,38 @@ func TestPendingPacketData(t *testing.T) {
 	}
 }
 
+// TestDeletePendingPacketDataForConsumer tests the DeletePendingPacketDataForConsumer method.
+func TestDeletePendingPacketDataForConsumer(t *testing.T) {
+
+	providerKeeper, ctx, ctrl, _ := testkeeper.GetProviderKeeperAndCtx(t, testkeeper.NewInMemKeeperParams(t))
+	defer ctrl.Finish()
+	providerKeeper.SetParams(ctx, providertypes.DefaultParams())
+
+	// Queue slash and a VSC matured packet data for chain-48
+	providerKeeper.QueuePendingSlashPacketData(ctx, "chain-48", 0, testkeeper.GetNewSlashPacketData())
+	providerKeeper.QueuePendingVSCMaturedPacketData(ctx, "chain-48", 1, testkeeper.GetNewVSCMaturedPacketData())
+
+	// Queue 3 slash, and 4 vsc matured packet data instances for chain-49
+	providerKeeper.QueuePendingSlashPacketData(ctx, "chain-49", 0, testkeeper.GetNewSlashPacketData())
+	providerKeeper.QueuePendingSlashPacketData(ctx, "chain-49", 1, testkeeper.GetNewSlashPacketData())
+	providerKeeper.QueuePendingSlashPacketData(ctx, "chain-49", 2, testkeeper.GetNewSlashPacketData())
+	providerKeeper.QueuePendingVSCMaturedPacketData(ctx, "chain-49", 3, testkeeper.GetNewVSCMaturedPacketData())
+	providerKeeper.QueuePendingVSCMaturedPacketData(ctx, "chain-49", 4, testkeeper.GetNewVSCMaturedPacketData())
+	providerKeeper.QueuePendingVSCMaturedPacketData(ctx, "chain-49", 5, testkeeper.GetNewVSCMaturedPacketData())
+	providerKeeper.QueuePendingVSCMaturedPacketData(ctx, "chain-49", 6, testkeeper.GetNewVSCMaturedPacketData())
+
+	// Delete all packet data for chain-49, confirm they are deleted
+	providerKeeper.DeletePendingPacketDataForConsumer(ctx, "chain-49")
+	slashData, vscMaturedData := providerKeeper.GetAllPendingPacketData(ctx, "chain-49")
+	require.Empty(t, slashData)
+	require.Empty(t, vscMaturedData)
+
+	// Confirm packet data for chain-48 is not deleted
+	slashData, vscMaturedData = providerKeeper.GetAllPendingPacketData(ctx, "chain-48")
+	require.Len(t, slashData, 1)
+	require.Len(t, vscMaturedData, 1)
+}
+
 // TestPanicIfTooMuchPendingPacketData tests the PanicIfTooMuchPendingPacketData method.
 func TestPanicIfTooMuchPendingPacketData(t *testing.T) {