feat!: throttle with retries provider changes

proto/interchain_security/ccv/provider/v1/query.proto

@@ -65,8 +65,7 @@
  "/interchain_security/ccv/provider/throttle_state";
  }
 
- // QueryThrottledConsumerPacketData returns a list of pending packet data
- // instances (slash packet and vsc matured) for a single consumer chain
+ // [DEPRECATED] Returns an empty set.
  rpc QueryThrottledConsumerPacketData(QueryThrottledConsumerPacketDataRequest)
  returns (QueryThrottledConsumerPacketDataResponse) {
  option (google.api.http).get =
@@ -86,7 +85,7 @@
 message QueryConsumerGenesisRequest { string chain_id = 1; }
 
 message QueryConsumerGenesisResponse {
  interchain_security.ccv.v1.GenesisState genesis_state = 1
  [ (gogoproto.nullable) = false ];
 }
 

tests/e2e/actions.go

@@ -1935,50 +1935,41 @@ func (tr TestRun) assignConsumerPubKey(action assignConsumerPubKeyAction, verbos
  tr.waitBlocks(chainID("provi"), 2, 30*time.Second)
 }
 
-// slashThrottleDequeue polls slash queue sizes until nextQueueSize is achieved
-type slashThrottleDequeue struct {
- chain chainID
- currentQueueSize int
- nextQueueSize int
+// slashMeterReplenishmentAction polls the slash meter on provider until value is achieved
+type slashMeterReplenishmentAction struct {
+ targetValue int64
  // panic if timeout is exceeded
  timeout time.Duration
 }
 
-func (tr TestRun) waitForSlashThrottleDequeue(
- action slashThrottleDequeue,
+func (tr TestRun) waitForSlashMeterReplenishment(
+ action slashMeterReplenishmentAction,
  verbose bool,
 ) {
  timeout := time.Now().Add(action.timeout)
- initialGlobalQueueSize := int(tr.getGlobalSlashQueueSize())
+ initialSlashMeter := tr.getSlashMeter()
 
- if initialGlobalQueueSize != action.currentQueueSize {
- panic(fmt.Sprintf("wrong initial queue size: %d - expected global queue: %d\n", initialGlobalQueueSize, action.currentQueueSize))
+ if initialSlashMeter >= 0 {
+ panic(fmt.Sprintf("No need to wait for slash meter replenishment, current value: %d", initialSlashMeter))
  }
+
  for {
- globalQueueSize := int(tr.getGlobalSlashQueueSize())
- chainQueueSize := int(tr.getConsumerChainPacketQueueSize(action.chain))
+ slashMeter := tr.getSlashMeter()
  if verbose {
- fmt.Printf("waiting for packed queue size to reach: %d - current: %d\n", action.nextQueueSize, globalQueueSize)
+ fmt.Printf("waiting for slash meter to be replenished, current value: %d\n", slashMeter)
  }
 
- // check if global queue size is equal to chain queue size
- if globalQueueSize == chainQueueSize && globalQueueSize == action.nextQueueSize { //nolint:gocritic // this is the comparison that we want here.
+ // check if meter has reached target value
+ if slashMeter >= action.targetValue {
  break
  }
 
  if time.Now().After(timeout) {
- panic(fmt.Sprintf("\n\n\nwaitForSlashThrottleDequeuemethod has timed out after: %s\n\n", action.timeout))
+ panic(fmt.Sprintf("\n\nwaitForSlashMeterReplenishment has timed out after: %s\n\n", action.timeout))
  }
 
  time.Sleep(500 * time.Millisecond)
  }
- // wair for 2 blocks to be created
- // allowing the jailing to be incorporated into voting power
- tr.waitBlocks(action.chain, 2, time.Minute)
-}
-
-func uintPointer(i uint) *uint {
- return &i
 }
 
 // GetPathNameForGorelayer returns the name of the path between two given chains used by Gorelayer.

tests/e2e/main.go

@@ -242,8 +242,8 @@ func (tr *TestRun) runStep(step Step, verbose bool) {
  tr.registerRepresentative(action, verbose)
  case assignConsumerPubKeyAction:
  tr.assignConsumerPubKey(action, verbose)
- case slashThrottleDequeue:
- tr.waitForSlashThrottleDequeue(action, verbose)
+ case slashMeterReplenishmentAction:
+ tr.waitForSlashMeterReplenishment(action, verbose)
  case startRelayerAction:
  tr.startRelayer(action, verbose)
  case registerConsumerRewardDenomAction:

tests/e2e/state.go

@@ -166,19 +166,6 @@ func (tr TestRun) getChainState(chain chainID, modelState ChainState) ChainState
  chainState.ProviderKeys = &providerKeys
  }
 
- if modelState.GlobalSlashQueueSize != nil {
- globalQueueSize := tr.getGlobalSlashQueueSize()
- chainState.GlobalSlashQueueSize = &globalQueueSize
- }
-
- if modelState.ConsumerChainQueueSizes != nil {
- consumerChainQueueSizes := map[chainID]uint{}
- for c := range *modelState.ConsumerChainQueueSizes {
- consumerChainQueueSizes[c] = tr.getConsumerChainPacketQueueSize(c)
- }
- chainState.ConsumerChainQueueSizes = &consumerChainQueueSizes
- }
-
  if modelState.RegisteredConsumerRewardDenoms != nil {
  registeredConsumerRewardDenoms := tr.getRegisteredConsumerRewardDenoms(chain)
  chainState.RegisteredConsumerRewardDenoms = &registeredConsumerRewardDenoms
@@ -657,9 +644,10 @@ func (tr TestRun) getProviderAddressFromConsumer(consumerChain chainID, validato
  return addr
 }
 
-func (tr TestRun) getGlobalSlashQueueSize() uint {
+func (tr TestRun) getSlashMeter() int64 {
  //#nosec G204 -- Bypass linter warning for spawning subprocess with cmd arguments.
- cmd := exec.Command("docker", "exec", tr.containerConfig.instanceName, tr.chainConfigs[chainID("provi")].binaryName,
+ cmd := exec.Command("docker", "exec",
+ tr.containerConfig.instanceName, tr.chainConfigs[chainID("provi")].binaryName,
 
  "query", "provider", "throttle-state",
  `--node`, tr.getQueryNode(chainID("provi")),
@@ -670,26 +658,8 @@ func (tr TestRun) getGlobalSlashQueueSize() uint {
  log.Fatal(err, "\n", string(bz))
  }
 
- packets := gjson.Get(string(bz), "packets").Array()
- return uint(len(packets))
-}
-
-func (tr TestRun) getConsumerChainPacketQueueSize(consumerChain chainID) uint {
- //#nosec G204 -- Bypass linter warning for spawning subprocess with cmd arguments.
- cmd := exec.Command("docker", "exec", tr.containerConfig.instanceName, tr.chainConfigs[chainID("provi")].binaryName,
-
- "query", "provider", "throttled-consumer-packet-data",
- string(consumerChain),
- `--node`, tr.getQueryNode(chainID("provi")),
- `-o`, `json`,
- )
- bz, err := cmd.CombinedOutput()
- if err != nil {
- log.Fatal(err, "\n", string(bz))
- }
-
- size := gjson.Get(string(bz), "size").Uint()
- return uint(size)
+ slashMeter := gjson.Get(string(bz), "slash_meter")
+ return slashMeter.Int()
 }
 
 func (tr TestRun) getRegisteredConsumerRewardDenoms(chain chainID) []string {

tests/e2e/steps_downtime.go

@@ -278,7 +278,7 @@ func stepsThrottledDowntime(consumerName string) []Step {
  validator: validatorID("bob"),
  },
  state: State{
- // slash packet queued on consumer, but powers not affected on either chain yet
+ // slash packet queued for bob on consumer, but powers not affected on either chain yet
  chainID("provi"): ChainState{
  ValPowers: &map[validatorID]uint{
  validatorID("alice"): 511,
@@ -312,11 +312,6 @@ func stepsThrottledDowntime(consumerName string) []Step {
  validatorID("bob"): 0, // bob is jailed
  validatorID("carol"): 500,
  },
- // no provider throttling engaged yet
- GlobalSlashQueueSize: uintPointer(0),
- ConsumerChainQueueSizes: &map[chainID]uint{
- chainID(consumerName): uint(0),
- },
  },
  chainID(consumerName): ChainState{
  // VSC packet applying jailing is not yet relayed to consumer
@@ -328,13 +323,13 @@ func stepsThrottledDowntime(consumerName string) []Step {
  },
  },
  },
+ // Invoke carol downtime slash on consumer
  {
  action: downtimeSlashAction{
  chain: chainID(consumerName),
  validator: validatorID("carol"),
  },
  state: State{
- // powers not affected on either chain yet
  chainID("provi"): ChainState{
  ValPowers: &map[validatorID]uint{
  validatorID("alice"): 511,
@@ -343,10 +338,9 @@ func stepsThrottledDowntime(consumerName string) []Step {
  },
  },
  chainID(consumerName): ChainState{
- // VSC packet applying jailing is not yet relayed to consumer
  ValPowers: &map[validatorID]uint{
  validatorID("alice"): 511,
- validatorID("bob"): 500,
+ validatorID("bob"): 500, // VSC packet applying bob jailing is not yet relayed to consumer
  validatorID("carol"): 500,
  },
  },
@@ -364,27 +358,24 @@ func stepsThrottledDowntime(consumerName string) []Step {
  ValPowers: &map[validatorID]uint{
  validatorID("alice"): 511,
  validatorID("bob"): 0,
- validatorID("carol"): 500, // not slashed due to throttling
- },
- GlobalSlashQueueSize: uintPointer(1), // carol's slash request is throttled
- ConsumerChainQueueSizes: &map[chainID]uint{
- chainID(consumerName): uint(1),
+ validatorID("carol"): 500, // slash packet for carol recv by provider, carol not slashed due to throttling
  },
  },
  chainID(consumerName): ChainState{
  ValPowers: &map[validatorID]uint{
  validatorID("alice"): 511,
- validatorID("bob"): 0,
+ validatorID("bob"): 0, // VSC packet applying bob jailing is also relayed and recv by consumer
  validatorID("carol"): 500,
  },
  },
  },
  },
+ // TODO(Shawn): Improve this test to have the consumer retry it's downtime slash, and to assert queue size on consumer.
+ // See https://github.com/cosmos/interchain-security/issues/1103 and https://github.com/cosmos/interchain-security/issues/1233
  {
- action: slashThrottleDequeue{
- chain: chainID(consumerName),
- currentQueueSize: 1,
- nextQueueSize: 0,
+ action: slashMeterReplenishmentAction{
+ targetValue: 0, // We just want slash meter to be non-negative
+
  // Slash meter replenish fraction is set to 10%, replenish period is 20 seconds, see config.go
  // Meter is initially at 10%, decremented to -23% from bob being jailed. It'll then take three replenishments
  // for meter to become positive again. 3*20 = 60 seconds + buffer = 80 seconds
@@ -395,11 +386,7 @@ func stepsThrottledDowntime(consumerName string) []Step {
  ValPowers: &map[validatorID]uint{
  validatorID("alice"): 511,
  validatorID("bob"): 0,
- validatorID("carol"): 0, // Carol is jailed upon packet being handled on provider
- },
- GlobalSlashQueueSize: uintPointer(0), // slash packets dequeued
- ConsumerChainQueueSizes: &map[chainID]uint{
- chainID(consumerName): 0,
+ validatorID("carol"): 500, // Carol still not slashed, packet must be retried
  },
  },
  chainID(consumerName): ChainState{
@@ -412,36 +399,5 @@ func stepsThrottledDowntime(consumerName string) []Step {
  },
  },
  },
- // A block is incremented each action, hence why VSC is committed on provider,
- // and can now be relayed as packet to consumer
- {
- action: relayPacketsAction{
- chainA: chainID("provi"),
- chainB: chainID(consumerName),
- port: "provider",
- channel: 0,
- },
- state: State{
- chainID("provi"): ChainState{
- ValPowers: &map[validatorID]uint{
- validatorID("alice"): 511,
- validatorID("bob"): 0,
- validatorID("carol"): 0,
- },
- GlobalSlashQueueSize: uintPointer(0),
- ConsumerChainQueueSizes: &map[chainID]uint{
- chainID(consumerName): 0,
- },
- },
- chainID(consumerName): ChainState{
- ValPowers: &map[validatorID]uint{
- validatorID("alice"): 511,
- // throttled update gets to consumer
- validatorID("bob"): 0,
- validatorID("carol"): 0,
- },
- },
- },
- },
  }
 }

tests/integration/common.go

@@ -1,6 +1,7 @@
 package integration
 
 import (
+ "bytes"
  "fmt"
  "time"
 
@@ -414,58 +415,41 @@ func (suite *CCVTestSuite) commitConsumerPacket(ctx sdk.Context, packetData ccv.
 func (s *CCVTestSuite) constructSlashPacketFromConsumer(bundle icstestingutils.ConsumerBundle,
  tmVal tmtypes.Validator, infractionType stakingtypes.Infraction, ibcSeqNum uint64,
 ) channeltypes.Packet {
+ packet, _ := s.constructSlashPacketFromConsumerWithData(bundle, tmVal, infractionType, ibcSeqNum)
+ return packet
+}
+
+func (s *CCVTestSuite) constructSlashPacketFromConsumerWithData(bundle icstestingutils.ConsumerBundle,
+ tmVal tmtypes.Validator, infractionType stakingtypes.Infraction, ibcSeqNum uint64,
+) (channeltypes.Packet, ccv.SlashPacketData) {
  valsetUpdateId := bundle.GetKeeper().GetHeightValsetUpdateID(
  bundle.GetCtx(), uint64(bundle.GetCtx().BlockHeight()))
 
- data := ccv.ConsumerPacketData{
- Type: ccv.SlashPacket,
- Data: &ccv.ConsumerPacketData_SlashPacketData{
- SlashPacketData: &ccv.SlashPacketData{
- Validator: abci.Validator{
- Address: tmVal.Address,
- Power: tmVal.VotingPower,
- },
- ValsetUpdateId: valsetUpdateId,
- Infraction: infractionType,
- },
+ spdData := ccv.SlashPacketData{
+ Validator: abci.Validator{
+ Address: tmVal.Address,
+ Power: tmVal.VotingPower,
  },
+ ValsetUpdateId: valsetUpdateId,
+ Infraction: infractionType,
  }
 
- return channeltypes.NewPacket(data.GetBytes(),
- ibcSeqNum,
- ccv.ConsumerPortID, // Src port
- bundle.Path.EndpointA.ChannelID, // Src channel
- ccv.ProviderPortID, // Dst port
- bundle.Path.EndpointB.ChannelID, // Dst channel
- clienttypes.Height{},
- uint64(bundle.GetCtx().BlockTime().Add(ccv.DefaultCCVTimeoutPeriod).UnixNano()),
- )
-}
-
-// constructVSCMaturedPacketFromConsumer constructs an IBC packet embedding
-// VSC Matured packet data to be sent from consumer to provider
-func (s *CCVTestSuite) constructVSCMaturedPacketFromConsumer(bundle icstestingutils.ConsumerBundle,
- ibcSeqNum uint64,
-) channeltypes.Packet {
- valsetUpdateId := bundle.GetKeeper().GetHeightValsetUpdateID(
- bundle.GetCtx(), uint64(bundle.GetCtx().BlockHeight()))
-
- data := ccv.ConsumerPacketData{
- Type: ccv.VscMaturedPacket,
- Data: &ccv.ConsumerPacketData_VscMaturedPacketData{
- VscMaturedPacketData: &ccv.VSCMaturedPacketData{ValsetUpdateId: valsetUpdateId},
+ cpdData := ccv.ConsumerPacketData{
+ Type: ccv.SlashPacket,
+ Data: &ccv.ConsumerPacketData_SlashPacketData{
+ SlashPacketData: &spdData,
  },
  }
 
- return channeltypes.NewPacket(data.GetBytes(),
+ return channeltypes.NewPacket(cpdData.GetBytes(),
  ibcSeqNum,
  ccv.ConsumerPortID, // Src port
  bundle.Path.EndpointA.ChannelID, // Src channel
  ccv.ProviderPortID, // Dst port
  bundle.Path.EndpointB.ChannelID, // Dst channel
  clienttypes.Height{},
  uint64(bundle.GetCtx().BlockTime().Add(ccv.DefaultCCVTimeoutPeriod).UnixNano()),
- )
+ ), spdData
 }
 
 // incrementTime increments the overall time by jumpPeriod
@@ -615,3 +599,18 @@ func (s *CCVTestSuite) setupValidatorPowers() {
  }
  s.Require().Equal(int64(4000), stakingKeeper.GetLastTotalPower(s.providerCtx()).Int64())
 }
+
+// mustGetStakingValFromTmVal returns the staking validator from the current state of the staking keeper,
+// corresponding to a given tendermint validator. Note this func will fail the test if the validator is not found.
+func (s *CCVTestSuite) mustGetStakingValFromTmVal(tmVal tmtypes.Validator) (stakingVal stakingtypes.Validator) {
+ vals := s.providerApp.GetTestStakingKeeper().GetAllValidators(s.providerCtx())
+ for i, val := range vals {
+ consAddr, err := val.GetConsAddr()
+ s.Require().NoError(err)
+ if bytes.Equal(consAddr.Bytes(), tmVal.Address.Bytes()) {
+ stakingVal = vals[i]
+ }
+ }
+ s.Require().NotZero(stakingVal)
+ return stakingVal
+}