fix: limit vsc matured packets handled per endblocker (#1004)

* initial implementation, still need tests

* UTs

* integration test

* linter

* Update CHANGELOG.md

* make vsc matured handled this block a var

* comment

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

@@ -53,14 +53,25 @@ func (k Keeper) OnRecvVSCMaturedPacket(
 // the "VSC Maturity and Slashing Order" CCV property. If VSC matured packet data DOES NOT
 // trail slash packet data for that consumer, it will be handled in this method,
 // bypassing HandleThrottleQueues.
-func (k Keeper) HandleLeadingVSCMaturedPackets(ctx sdk.Context) {
+func (k Keeper) HandleLeadingVSCMaturedPackets(ctx sdk.Context) (vscMaturedHandledThisBlock int) {
+	vscMaturedHandledThisBlock = 0
 	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 vscMaturedHandledThisBlock >= vscMaturedHandledPerBlockLimit {
+				// Break from inner for-loop, DeleteThrottledPacketData will still be called for this consumer
+				break
+			}
 			k.HandleVSCMaturedPacket(ctx, chain.ChainId, data)
+			vscMaturedHandledThisBlock++
+			ibcSeqNumsHandled = append(ibcSeqNumsHandled, ibcSeqNums[idx])
 		}
-		k.DeleteThrottledPacketData(ctx, chain.ChainId, ibcSeqNums...)
+		k.DeleteThrottledPacketData(ctx, chain.ChainId, ibcSeqNumsHandled...)
 	}
+	return vscMaturedHandledThisBlock
 }
 
 // HandleVSCMaturedPacket handles a VSCMatured packet.
@@ -267,13 +278,14 @@ func (k Keeper) EndBlockCIS(ctx sdk.Context) {
 	// - Marshaling and/or store corruption errors.
 	// - Setting invalid slash meter values (see SetSlashMeter).
 	k.CheckForSlashMeterReplenishment(ctx)
+
 	// Handle leading vsc matured packets before throttling logic.
 	//
 	// Note: HandleLeadingVSCMaturedPackets contains panics for the following scenarios, any of which should never occur
 	// if the protocol is correct and external data is properly validated:
 	//
 	// - Marshaling and/or store corruption errors.
-	k.HandleLeadingVSCMaturedPackets(ctx)
+	vscMaturedHandledThisBlock := k.HandleLeadingVSCMaturedPackets(ctx)
 	// Handle queue entries considering throttling logic.
 	//
 	// Note: HandleThrottleQueues contains panics for the following scenarios, any of which should never occur
@@ -282,7 +294,7 @@ func (k Keeper) EndBlockCIS(ctx sdk.Context) {
 	// - SlashMeter has not been set (which should be set in InitGenesis, see InitializeSlashMeter).
 	// - Marshaling and/or store corruption errors.
 	// - Setting invalid slash meter values (see SetSlashMeter).
-	k.HandleThrottleQueues(ctx)
+	k.HandleThrottleQueues(ctx, vscMaturedHandledThisBlock)
 }
 
 // OnRecvSlashPacket delivers a received slash packet, validates it and

x/ccv/provider/keeper/throttle.go

@@ -12,16 +12,25 @@ import (
 
 // This file contains functionality relevant to the throttling of slash and vsc matured packets, aka circuit breaker logic.
 
+const vscMaturedHandledPerBlockLimit = 100
+
 // HandleThrottleQueues iterates over the global slash entry queue, and
 // handles all or some portion of throttled (slash and/or VSC matured) packet data received from
 // consumer chains. The slash meter is decremented appropriately in this method.
-func (k Keeper) HandleThrottleQueues(ctx sdktypes.Context) {
+func (k Keeper) HandleThrottleQueues(ctx sdktypes.Context, vscMaturedHandledThisBlock int) {
 	meter := k.GetSlashMeter(ctx)
 	// Return if meter is negative in value
 	if meter.IsNegative() {
 		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 vscMaturedHandledThisBlock >= vscMaturedHandledPerBlockLimit {
+		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)
@@ -35,7 +44,7 @@ func (k Keeper) HandleThrottleQueues(ctx sdktypes.Context) {
 		// Handle one slash and any trailing vsc matured packet data instances by passing in
 		// chainID and appropriate callbacks, relevant packet data is deleted in this method.
 
-		k.HandlePacketDataForChain(ctx, globalEntry.ConsumerChainID, k.HandleSlashPacket, k.HandleVSCMaturedPacket)
+		k.HandlePacketDataForChain(ctx, globalEntry.ConsumerChainID, k.HandleSlashPacket, k.HandleVSCMaturedPacket, vscMaturedHandledThisBlock)
 		handledEntries = append(handledEntries, globalEntry)
 
 		// don't handle any more global entries if meter becomes negative in value
@@ -82,18 +91,31 @@ func (k Keeper) GetEffectiveValPower(ctx sdktypes.Context,
 func (k Keeper) HandlePacketDataForChain(ctx sdktypes.Context, consumerChainID string,
 	slashPacketHandler func(sdktypes.Context, string, ccvtypes.SlashPacketData),
 	vscMaturedPacketHandler func(sdktypes.Context, string, ccvtypes.VSCMaturedPacketData),
+	vscMaturedHandledThisBlock int,
 ) {
 	// 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)
+		// Due to HandleLeadingVSCMaturedPackets() running before HandleThrottleQueues(), and the fact that
+		// HandleThrottleQueues() will return until all leading vsc matured have been handled, a slash packet
+		// should always be the first packet in the queue. So we can safely append the first seqNum here.
+		seqNumsHandled = append(seqNumsHandled, seqNums[0])
 	}
-	for _, vscMData := range vscMaturedData {
+	for idx, vscMData := range vscMaturedData {
+		if vscMaturedHandledThisBlock >= vscMaturedHandledPerBlockLimit {
+			// Break from for-loop, DeleteThrottledPacketData will still be called for this consumer
+			break
+		}
 		vscMaturedPacketHandler(ctx, consumerChainID, vscMData)
+		vscMaturedHandledThisBlock++
+		// 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),

x/ccv/provider/keeper/throttle_test.go

@@ -134,7 +134,7 @@ func TestHandlePacketDataForChain(t *testing.T) {
 			handledData = append(handledData, data)
 		}
 
-		providerKeeper.HandlePacketDataForChain(ctx, tc.chainID, slashHandleCounter, vscMaturedHandleCounter)
+		providerKeeper.HandlePacketDataForChain(ctx, tc.chainID, slashHandleCounter, vscMaturedHandleCounter, 0)
 
 		// Assert number of handled data instances matches expected number
 		require.Equal(t, len(tc.expectedHandledIndexes), len(handledData))

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(),
 	}
 }