fix: limit vsc matured packets handled per endblocker

CHANGELOG.md

@@ -25,6 +25,7 @@ Some PRs from v2.0.0 may reappear from other releases below. This is due to the
 
 ## Notable PRs included in v2.0.0
 
+* (feat/fix) limit vsc matured packets handled per endblocker [#1004](https://github.com/cosmos/interchain-security/pull/1004)
 * (fix) cosumer key prefix order to avoid complex migrations [#963](https://github.com/cosmos/interchain-security/pull/963) and [#991](https://github.com/cosmos/interchain-security/pull/991). The latter PR is the proper fix.
 * (feat) v1->v2 migrations to accommodate a bugfix having to do with store keys, introduce new params, and deal with consumer genesis state schema changes [#975](https://github.com/cosmos/interchain-security/pull/975) and [#997](https://github.com/cosmos/interchain-security/pull/997)
 * (deps) Bump github.com/cosmos/ibc-go/v4 from 4.4.0 to 4.4.2 [#982](https://github.com/cosmos/interchain-security/pull/982)

tests/integration/throttle.go

@@ -12,6 +12,8 @@ import (
 	tmtypes "github.com/tendermint/tendermint/types"
 )
 
+const fullSlashMeterString = "1.0"
+
 // TestBasicSlashPacketThrottling tests slash packet throttling with a single consumer,
 // two slash packets, and no VSC matured packets. The most basic scenario.
 func (s *CCVTestSuite) TestBasicSlashPacketThrottling() {
@@ -651,7 +653,7 @@ func (s *CCVTestSuite) TestSlashSameValidator() {
 
 	// Set replenish fraction to 1.0 so that all sent packets should handled immediately (no throttling)
 	params := providerKeeper.GetParams(s.providerCtx())
-	params.SlashMeterReplenishFraction = "1.0"
+	params.SlashMeterReplenishFraction = fullSlashMeterString // needs to be const for linter
 	providerKeeper.SetParams(s.providerCtx(), params)
 	providerKeeper.InitializeSlashMeter(s.providerCtx())
 
@@ -706,7 +708,7 @@ func (s CCVTestSuite) TestSlashAllValidators() { //nolint:govet // this is a tes
 
 	// Set replenish fraction to 1.0 so that all sent packets should be handled immediately (no throttling)
 	params := providerKeeper.GetParams(s.providerCtx())
-	params.SlashMeterReplenishFraction = "1.0"
+	params.SlashMeterReplenishFraction = fullSlashMeterString // needs to be const for linter
 	providerKeeper.SetParams(s.providerCtx(), params)
 	providerKeeper.InitializeSlashMeter(s.providerCtx())
 
@@ -779,7 +781,7 @@ func (s *CCVTestSuite) TestLeadingVSCMaturedAreDequeued() {
 
 	// Queue up 50 slash packets for each consumer
 	for _, bundle := range s.consumerBundles {
-		for i := 0; i < 50; i++ {
+		for i := 50; i < 100; i++ {
 			ibcSeqNum := uint64(i)
 			packet := s.constructSlashPacketFromConsumer(*bundle,
 				*s.providerChain.Vals.Validators[0], stakingtypes.Downtime, ibcSeqNum)
@@ -792,7 +794,7 @@ func (s *CCVTestSuite) TestLeadingVSCMaturedAreDequeued() {
 
 	// Queue up another 50 vsc matured packets for each consumer
 	for _, bundle := range s.consumerBundles {
-		for i := 0; i < 50; i++ {
+		for i := 100; i < 150; i++ {
 			ibcSeqNum := uint64(i)
 			packet := s.constructVSCMaturedPacketFromConsumer(*bundle, ibcSeqNum)
 			packetData := ccvtypes.ConsumerPacketData{}
@@ -818,6 +820,10 @@ func (s *CCVTestSuite) TestLeadingVSCMaturedAreDequeued() {
 	providerKeeper.SetSlashMeterReplenishTimeCandidate(s.providerCtx())
 
 	// Execute end blocker to dequeue only the leading vsc matured packets.
+	// Note we must call the end blocker three times, since only 100 vsc matured packets can be handled
+	// each block, and we have 5*50=250 total.
+	s.providerChain.NextBlock()
+	s.providerChain.NextBlock()
 	s.providerChain.NextBlock()
 
 	// Confirm queue size is 100 for each consumer-specific queue (50 leading vsc matured are dequeued).
@@ -827,9 +833,80 @@ func (s *CCVTestSuite) TestLeadingVSCMaturedAreDequeued() {
 	}
 
 	// No slash packets handled, global slash queue is same size as last block.
+	globalEntries = providerKeeper.GetAllGlobalSlashEntries(s.providerCtx())
+	s.Require().Equal(len(globalEntries), 50*5)
+
+	// No slash packets handled even if we call end blocker a couple more times.
+	s.providerChain.NextBlock()
+	s.providerChain.NextBlock()
+	globalEntries = providerKeeper.GetAllGlobalSlashEntries(s.providerCtx())
 	s.Require().Equal(len(globalEntries), 50*5)
 }
 
+// TestVscMaturedHandledPerBlockLimit tests that only 100 vsc matured packets are handled per block,
+// specifically from HandleThrottleQueues().
+//
+// Note the vsc matured per block limit is also tested in, TestLeadingVSCMaturedAreDequeued,
+// specifically in the context of HandleLeadingVSCMaturedPackets().
+func (s *CCVTestSuite) TestVscMaturedHandledPerBlockLimit() {
+	s.SetupAllCCVChannels()
+	providerKeeper := s.providerApp.GetProviderKeeper()
+
+	// Set replenish fraction to 1.0 so that all sent packets should be handled immediately (no jail throttling)
+	params := providerKeeper.GetParams(s.providerCtx())
+	params.SlashMeterReplenishFraction = fullSlashMeterString // needs to be const for linter
+	providerKeeper.SetParams(s.providerCtx(), params)
+	providerKeeper.InitializeSlashMeter(s.providerCtx())
+
+	// Queue up 100 vsc matured packets for each consumer
+	for _, bundle := range s.consumerBundles {
+		for i := 0; i < 100; i++ {
+			ibcSeqNum := uint64(i)
+			packet := s.constructVSCMaturedPacketFromConsumer(*bundle, ibcSeqNum)
+			packetData := ccvtypes.ConsumerPacketData{}
+			ccvtypes.ModuleCdc.MustUnmarshalJSON(packet.GetData(), &packetData)
+			providerKeeper.OnRecvVSCMaturedPacket(s.providerCtx(),
+				packet, *packetData.GetVscMaturedPacketData())
+		}
+	}
+
+	// Queue up 50 slash packets for each consumer, with new IBC sequence numbers
+	for _, bundle := range s.consumerBundles {
+		for i := 100; i < 150; i++ {
+			ibcSeqNum := uint64(i)
+			packet := s.constructSlashPacketFromConsumer(*bundle,
+				*s.providerChain.Vals.Validators[0], stakingtypes.Downtime, ibcSeqNum)
+			packetData := ccvtypes.ConsumerPacketData{}
+			ccvtypes.ModuleCdc.MustUnmarshalJSON(packet.GetData(), &packetData)
+			providerKeeper.OnRecvSlashPacket(s.providerCtx(),
+				packet, *packetData.GetSlashPacketData())
+		}
+	}
+
+	// Confirm queue size is 150 for each consumer-specific queue.
+	for _, bundle := range s.consumerBundles {
+		s.Require().Equal(uint64(150),
+			providerKeeper.GetThrottledPacketDataSize(s.providerCtx(), bundle.Chain.ChainID))
+	}
+	// Confirm global queue size is 50 * 5 (50 slash packets for each of 5 consumers)
+	globalEntries := providerKeeper.GetAllGlobalSlashEntries(s.providerCtx())
+	s.Require().Equal(len(globalEntries), 50*5)
+
+	// Note even though there is no jail throttling active, slash packets will not be handled until
+	// all of the leading vsc matured packets are handled first. This should take 5 blocks.
+	for i := 0; i < 5; i++ {
+		s.providerChain.NextBlock()
+		s.Require().Len(providerKeeper.GetAllGlobalSlashEntries(s.providerCtx()), 250) // global entries remain same size
+	}
+
+	// Set signing info for val to be jailed, preventing panic
+	s.setDefaultValSigningInfo(*s.providerChain.Vals.Validators[0])
+
+	// Now we execute one more block and all 250 of the slash packets should be handled.
+	s.providerChain.NextBlock()
+	s.Require().Empty(providerKeeper.GetAllGlobalSlashEntries(s.providerCtx())) // empty global entries = all slash packets handled
+}
+
 func (s *CCVTestSuite) confirmValidatorJailed(tmVal tmtypes.Validator, checkPower bool) {
 	sdkVal, found := s.providerApp.GetTestStakingKeeper().GetValidator(
 		s.providerCtx(), sdk.ValAddress(tmVal.Address))

testutil/integration/debug_test.go

@@ -205,6 +205,10 @@ func TestLeadingVSCMaturedAreDequeued(t *testing.T) {
 	runCCVTestByName(t, "TestLeadingVSCMaturedAreDequeued")
 }
 
+func TestVscMaturedHandledPerBlockLimit(t *testing.T) {
+	runCCVTestByName(t, "TestVscMaturedHandledPerBlockLimit")
+}
+
 //
 // Unbonding tests
 //

x/ccv/provider/keeper/relay.go

@@ -55,11 +55,19 @@ func (k Keeper) OnRecvVSCMaturedPacket(
 // bypassing HandleThrottleQueues.
 func (k Keeper) HandleLeadingVSCMaturedPackets(ctx sdk.Context) {
 	for _, chain := range k.GetAllConsumerChains(ctx) {
+		// Note: it's assumed the order of the vsc matured slice matches the order of the ibc seq nums slice,
+		// in that a vsc matured packet data at index i is associated with the ibc seq num at index i.
 		leadingVscMatured, ibcSeqNums := k.GetLeadingVSCMaturedData(ctx, chain.ChainId)
-		for _, data := range leadingVscMatured {
+		ibcSeqNumsHandled := []uint64{}
+		for idx, data := range leadingVscMatured {
+			if k.VSCMaturedHandledLimitReached(ctx) {
+				// Break from inner for-loop, DeleteThrottledPacketData will still be called for this consumer
+				break
+			}
 			k.HandleVSCMaturedPacket(ctx, chain.ChainId, data)
+			ibcSeqNumsHandled = append(ibcSeqNumsHandled, ibcSeqNums[idx])
 		}
-		k.DeleteThrottledPacketData(ctx, chain.ChainId, ibcSeqNums...)
+		k.DeleteThrottledPacketData(ctx, chain.ChainId, ibcSeqNumsHandled...)
 	}
 }
 
@@ -95,6 +103,9 @@ func (k Keeper) HandleVSCMaturedPacket(ctx sdk.Context, chainID string, data ccv
 		"chainID", chainID,
 		"vscID", data.ValsetUpdateId,
 	)
+
+	// Keep track of how many vsc matured packets have been handled this block
+	k.IncrementVSCMaturedHandledThisBlock(ctx)
 }
 
 // CompleteMaturedUnbondingOps attempts to complete all matured unbonding operations
@@ -267,6 +278,11 @@ func (k Keeper) EndBlockCIS(ctx sdk.Context) {
 	// - Marshaling and/or store corruption errors.
 	// - Setting invalid slash meter values (see SetSlashMeter).
 	k.CheckForSlashMeterReplenishment(ctx)
+
+	// Reset the vsc matured packet handler counter.
+	// Note this assumes vsc matured packets are only handled in the two methods following this call.
+	k.ResetVSCMaturedHandledThisBlock(ctx)
+
 	// Handle leading vsc matured packets before throttling logic.
 	//
 	// Note: HandleLeadingVSCMaturedPackets contains panics for the following scenarios, any of which should never occur

x/ccv/provider/keeper/throttle.go

@@ -22,6 +22,13 @@ func (k Keeper) HandleThrottleQueues(ctx sdktypes.Context) {
 		return
 	}
 
+	// Return if vsc matured handle limit was already reached this block, during HandleLeadingVSCMaturedPackets.
+	// It makes no practical difference for throttling logic to execute next block.
+	// By doing this, we assure that all leading vsc matured packets were handled before any slash packets.
+	if k.VSCMaturedHandledLimitReached(ctx) {
+		return
+	}
+
 	// Obtain all global slash entries, where only some of them may be handled in this method,
 	// depending on the value of the slash meter.
 	allEntries := k.GetAllGlobalSlashEntries(ctx)
@@ -85,15 +92,24 @@ func (k Keeper) HandlePacketDataForChain(ctx sdktypes.Context, consumerChainID s
 ) {
 	// Get slash packet data and trailing vsc matured packet data, handle it all.
 	slashFound, slashData, vscMaturedData, seqNums := k.GetSlashAndTrailingData(ctx, consumerChainID)
+	seqNumsHandled := []uint64{}
 	if slashFound {
 		slashPacketHandler(ctx, consumerChainID, slashData)
+		// Slash packet should always be the first packet in the queue, so we can safely append the first seqNum.
+		seqNumsHandled = append(seqNumsHandled, seqNums[0])
 	}
-	for _, vscMData := range vscMaturedData {
+	for idx, vscMData := range vscMaturedData {
+		if k.VSCMaturedHandledLimitReached(ctx) {
+			// Break from for-loop, DeleteThrottledPacketData will still be called for this consumer
+			break
+		}
 		vscMaturedPacketHandler(ctx, consumerChainID, vscMData)
+		// Append seq num for this vsc matured packet
+		seqNumsHandled = append(seqNumsHandled, seqNums[idx+1]) // Note idx+1, since slash packet is at index 0
 	}
 
 	// Delete handled data after it has all been handled.
-	k.DeleteThrottledPacketData(ctx, consumerChainID, seqNums...)
+	k.DeleteThrottledPacketData(ctx, consumerChainID, seqNumsHandled...)
 }
 
 // InitializeSlashMeter initializes the slash meter to it's max value (also its allowance),
@@ -607,3 +623,27 @@ func (k Keeper) SetSlashMeterReplenishTimeCandidate(ctx sdktypes.Context) {
 	timeToStore := ctx.BlockTime().UTC().Add(k.GetSlashMeterReplenishPeriod(ctx))
 	store.Set(providertypes.SlashMeterReplenishTimeCandidateKey(), sdktypes.FormatTimeBytes(timeToStore))
 }
+
+func (k Keeper) IncrementVSCMaturedHandledThisBlock(ctx sdktypes.Context) {
+	current := k.GetVSCMaturedHandledThisBlock(ctx)
+	store := ctx.KVStore(k.storeKey)
+	store.Set(providertypes.VSCMaturedHandledThisBlockKey(), sdktypes.Uint64ToBigEndian(current+1))
+}
+
+func (k Keeper) VSCMaturedHandledLimitReached(ctx sdktypes.Context) bool {
+	return k.GetVSCMaturedHandledThisBlock(ctx) >= 100
+}
+
+func (k Keeper) GetVSCMaturedHandledThisBlock(ctx sdktypes.Context) uint64 {
+	store := ctx.KVStore(k.storeKey)
+	bz := store.Get(providertypes.VSCMaturedHandledThisBlockKey())
+	if bz == nil {
+		return 0
+	}
+	return sdktypes.BigEndianToUint64(bz)
+}
+
+func (k Keeper) ResetVSCMaturedHandledThisBlock(ctx sdktypes.Context) {
+	store := ctx.KVStore(k.storeKey)
+	store.Delete(providertypes.VSCMaturedHandledThisBlockKey())
+}

x/ccv/provider/keeper/throttle_test.go

@@ -1315,6 +1315,46 @@ func TestSlashMeterReplenishTimeCandidate(t *testing.T) {
 	}
 }
 
+func TestVscMaturedHandledThisBlockCRUD(t *testing.T) {
+	providerKeeper, ctx, ctrl, _ := testkeeper.GetProviderKeeperAndCtx(t, testkeeper.NewInMemKeeperParams(t))
+	defer ctrl.Finish()
+
+	// Confirm initial value is 0
+	require.Equal(t, uint64(0), providerKeeper.GetVSCMaturedHandledThisBlock(ctx))
+	require.False(t, providerKeeper.VSCMaturedHandledLimitReached(ctx)) // limit is always 100
+
+	// Increment 25 times and confirm values
+	for i := uint64(0); i < 25; i++ {
+		before := providerKeeper.GetVSCMaturedHandledThisBlock(ctx)
+		providerKeeper.IncrementVSCMaturedHandledThisBlock(ctx)
+		after := providerKeeper.GetVSCMaturedHandledThisBlock(ctx)
+		require.Equal(t, before+1, after)
+		require.False(t, providerKeeper.VSCMaturedHandledLimitReached(ctx))
+	}
+	require.Equal(t, uint64(25), providerKeeper.GetVSCMaturedHandledThisBlock(ctx))
+
+	// Confirm that resetting works
+	providerKeeper.ResetVSCMaturedHandledThisBlock(ctx)
+	require.Equal(t, uint64(0), providerKeeper.GetVSCMaturedHandledThisBlock(ctx))
+
+	// Increment 99 times
+	for i := uint64(0); i < 99; i++ {
+		providerKeeper.IncrementVSCMaturedHandledThisBlock(ctx)
+	}
+
+	// Confirm limit not reached
+	require.False(t, providerKeeper.VSCMaturedHandledLimitReached(ctx))
+
+	// Increment more, confirm that limit is reached
+	providerKeeper.IncrementVSCMaturedHandledThisBlock(ctx)
+	require.Equal(t, uint64(100), providerKeeper.GetVSCMaturedHandledThisBlock(ctx))
+	require.True(t, providerKeeper.VSCMaturedHandledLimitReached(ctx))
+
+	providerKeeper.IncrementVSCMaturedHandledThisBlock(ctx)
+	require.Equal(t, uint64(101), providerKeeper.GetVSCMaturedHandledThisBlock(ctx))
+	require.True(t, providerKeeper.VSCMaturedHandledLimitReached(ctx))
+}
+
 // Struct used for TestPendingPacketData and helpers
 type throttledPacketDataInstance struct {
 	IbcSeqNum uint64

x/ccv/provider/types/keys.go

@@ -134,6 +134,10 @@ const (
 	// ConsumerRewardDenomsBytePrefix is the byte prefix that will store a list of consumer reward denoms
 	ConsumerRewardDenomsBytePrefix
 
+	// VSCMaturedHandledThisBlockBytePrefix is the byte prefix storing the number of vsc matured packets
+	// handled in the current block
+	VSCMaturedHandledThisBlockBytePrefix
+
 	// NOTE: DO NOT ADD NEW BYTE PREFIXES HERE WITHOUT ADDING THEM TO getAllKeyPrefixes() IN keys_test.go
 )
 
@@ -476,6 +480,10 @@ func ParseChainIdAndConsAddrKey(prefix byte, bz []byte) (string, sdk.ConsAddress
 	return chainID, addr, nil
 }
 
+func VSCMaturedHandledThisBlockKey() []byte {
+	return []byte{VSCMaturedHandledThisBlockBytePrefix}
+}
+
 //
 // End of generic helpers section
 //

x/ccv/provider/types/keys_test.go

@@ -51,6 +51,7 @@ func getAllKeyPrefixes() []byte {
 		providertypes.KeyAssignmentReplacementsBytePrefix,
 		providertypes.ConsumerAddrsToPruneBytePrefix,
 		providertypes.SlashLogBytePrefix,
+		providertypes.VSCMaturedHandledThisBlockBytePrefix,
 	}
 }
 
@@ -94,6 +95,7 @@ func getAllFullyDefinedKeys() [][]byte {
 		providertypes.KeyAssignmentReplacementsKey("chainID", providertypes.NewProviderConsAddress([]byte{0x05})),
 		providertypes.ConsumerAddrsToPruneKey("chainID", 88),
 		providertypes.SlashLogKey(providertypes.NewProviderConsAddress([]byte{0x05})),
+		providertypes.VSCMaturedHandledThisBlockKey(),
 	}
 }