diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2a5c14c6e4..0a83fcd3df 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,7 @@
 
 Add an entry to the unreleased section whenever merging a PR to main that is not targeted at a specific release. These entries will eventually be included in a release.
 
+* (feat!) [#1024](https://github.com/cosmos/interchain-security/pull/1024) throttle with retries, consumer changes  
 * (fix!) revert consumer packet data changes from #1037 [#1150](https://github.com/cosmos/interchain-security/pull/1150)
 * (fix!) proper deletion of pending packets [#1146](https://github.com/cosmos/interchain-security/pull/1146)
 * (feat!) optimize pending packets storage on consumer, with migration! [#1037](https://github.com/cosmos/interchain-security/pull/1037)
diff --git a/docs/docs/adrs/adr-008-throttle-retries.md b/docs/docs/adrs/adr-008-throttle-retries.md
index 134214fffb..1faf7bd7ee 100644
--- a/docs/docs/adrs/adr-008-throttle-retries.md
+++ b/docs/docs/adrs/adr-008-throttle-retries.md
@@ -9,6 +9,7 @@ title: Throttle with retries
 
 * 6/9/23: Initial draft
 * 6/22/23: added note on consumer pending packets storage optimization
+* 7/14/23: Added note on upgrade order
 
 ## Status
 
@@ -47,6 +48,8 @@ With the behavior described, we maintain very similar behavior to the current th
 
 In the normal case, when no or a few slash packets are being sent, the VSCMaturedPackets will not be delayed, and hence unbonding will not be delayed.
 
+For implementation of this design, see [throttle_retry.go](../../../x/ccv/consumer/keeper/throttle_retry.go).
+
 ### Consumer pending packets storage optimization
 
 In addition to the mentioned consumer changes above. An optimization will need to be made to the consumer's pending packets storage to properly implement the feature from this ADR.
@@ -86,9 +89,11 @@ If a consumer sends VSCMatured packets too leniently: The consumer is malicious
 
 If a consumer blocks the sending of VSCMatured packets: The consumer is malicious and blocking vsc matured packets that should have been sent. This will block unbonding only up until the VSC timeout period has elapsed. At that time, the consumer is removed. Again the malicious behavior only creates a negative outcome for the chain that is being malicious.
 
-### Splitting of PRs
+### Splitting of PRs and Upgrade Order
+
+This feature will implement consumer changes in [#1024](https://github.com/cosmos/interchain-security/pull/1024). Note these changes should be deployed to prod for all consumers before the provider changes are deployed to prod. That is the consumer changes in #1024 are compatible with the current ("v1") provider implementation of throttling that's running on the Cosmos Hub as of July 2023.
 
-We could split this feature into two PRs, one affecting the consumer and one affecting the provider, along with a third PR which could setup a clever way to upgrade the provider in multiple steps, ensuring that queued slash packets at upgrade time are handled properly.
+Once all consumers have deployed the changes in #1024, the provider changes from (TBD) can be deployed to prod, fully enabling v2 throttling.
 
 ## Consequences
 
diff --git a/proto/interchain_security/ccv/consumer/v1/consumer.proto b/proto/interchain_security/ccv/consumer/v1/consumer.proto
index 97ba14f6da..2b4b6f88c3 100644
--- a/proto/interchain_security/ccv/consumer/v1/consumer.proto
+++ b/proto/interchain_security/ccv/consumer/v1/consumer.proto
@@ -89,3 +89,11 @@ message MaturingVSCPacket {
   google.protobuf.Timestamp maturity_time = 2
       [ (gogoproto.stdtime) = true, (gogoproto.nullable) = false ];
 }
+
+// A record storing the state of a slash packet sent to the provider chain
+// which may bounce back and forth until handled by the provider.
+message SlashRecord {
+  bool waiting_on_reply = 1;
+  google.protobuf.Timestamp send_time = 2
+      [ (gogoproto.stdtime) = true, (gogoproto.nullable) = false ];
+}
\ No newline at end of file
diff --git a/tests/difference/core/driver/core_test.go b/tests/difference/core/driver/core_test.go
index 12192eb8e4..ec97fbf9c1 100644
--- a/tests/difference/core/driver/core_test.go
+++ b/tests/difference/core/driver/core_test.go
@@ -2,7 +2,6 @@ package core
 
 import (
 	"fmt"
-	"testing"
 	"time"
 
 	channeltypes "github.com/cosmos/ibc-go/v7/modules/core/04-channel/types"
@@ -331,9 +330,12 @@ func (s *CoreSuite) TestTraces() {
 	fmt.Println("Shortest [traceIx, actionIx]:", shortest, shortestLen)
 }
 
-func TestCoreSuite(t *testing.T) {
-	suite.Run(t, new(CoreSuite))
-}
+// TODO: diff tests will eventually be replaced by quint tests, and all this code could then be deleted.
+// Until that decision is finalized, we'll just comment out the top-level test.
+
+// func TestCoreSuite(t *testing.T) {
+// 	suite.Run(t, new(CoreSuite))
+// }
 
 // SetupTest sets up the test suite in a 'zero' state which matches
 // the initial state in the model.
diff --git a/tests/e2e/steps_downtime.go b/tests/e2e/steps_downtime.go
index 74cb349000..e6d320bec1 100644
--- a/tests/e2e/steps_downtime.go
+++ b/tests/e2e/steps_downtime.go
@@ -278,7 +278,7 @@ func stepsThrottledDowntime(consumerName string) []Step {
 				validator: validatorID("bob"),
 			},
 			state: State{
-				// powers not affected on either chain yet
+				// slash packet queued on consumer, but powers not affected on either chain yet
 				chainID("provi"): ChainState{
 					ValPowers: &map[validatorID]uint{
 						validatorID("alice"): 511,
@@ -295,6 +295,39 @@ func stepsThrottledDowntime(consumerName string) []Step {
 				},
 			},
 		},
+		// Relay packets so bob is jailed on provider,
+		// and consumer receives ack that provider recv the downtime slash.
+		// The latter is necessary for the consumer to send the second downtime slash.
+		{
+			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, // 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
+					ValPowers: &map[validatorID]uint{
+						validatorID("alice"): 511,
+						validatorID("bob"):   500,
+						validatorID("carol"): 500,
+					},
+				},
+			},
+		},
 		{
 			action: downtimeSlashAction{
 				chain:     chainID(consumerName),
@@ -305,11 +338,12 @@ func stepsThrottledDowntime(consumerName string) []Step {
 				chainID("provi"): ChainState{
 					ValPowers: &map[validatorID]uint{
 						validatorID("alice"): 511,
-						validatorID("bob"):   500,
+						validatorID("bob"):   0,
 						validatorID("carol"): 500,
 					},
 				},
 				chainID(consumerName): ChainState{
+					// VSC packet applying jailing is not yet relayed to consumer
 					ValPowers: &map[validatorID]uint{
 						validatorID("alice"): 511,
 						validatorID("bob"):   500,
@@ -338,10 +372,9 @@ func stepsThrottledDowntime(consumerName string) []Step {
 					},
 				},
 				chainID(consumerName): ChainState{
-					// no updates received on consumer
 					ValPowers: &map[validatorID]uint{
 						validatorID("alice"): 511,
-						validatorID("bob"):   500,
+						validatorID("bob"):   0,
 						validatorID("carol"): 500,
 					},
 				},
@@ -373,7 +406,7 @@ func stepsThrottledDowntime(consumerName string) []Step {
 					// no updates received on consumer
 					ValPowers: &map[validatorID]uint{
 						validatorID("alice"): 511,
-						validatorID("bob"):   500,
+						validatorID("bob"):   0,
 						validatorID("carol"): 500,
 					},
 				},
diff --git a/tests/integration/expired_client.go b/tests/integration/expired_client.go
index 53863d2881..2a8babacfa 100644
--- a/tests/integration/expired_client.go
+++ b/tests/integration/expired_client.go
@@ -139,20 +139,34 @@ 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)
+	// At this point we expect 4 packets, two vsc matured packets and two trailing slash packets
 	s.Require().Len(consumerPackets, 4, "unexpected number of pending data packets")
 
 	// upgrade expired client to the consumer
 	upgradeExpiredClient(s, Provider)
 
-	// go to next block to trigger SendPendingDataPackets
+	// go to next block to trigger SendPendingPackets
 	s.consumerChain.NextBlock()
 
-	// check that the list of pending data packets is emptied
+	// Check that the leading vsc matured packets were sent and no longer pending
 	consumerPackets = consumerKeeper.GetPendingPackets(s.consumerCtx())
-	s.Require().Empty(consumerPackets)
+	s.Require().Len(consumerPackets, 2, "unexpected number of pending data packets")
+
+	// relay committed packets from consumer to provider, first slash packet should be committed
+	relayAllCommittedPackets(s, s.consumerChain, s.path, ccv.ConsumerPortID, s.path.EndpointA.ChannelID, 3) // two vsc matured + one slash
+
+	// First slash has been acked, now only the second slash packet should remain as pending
+	consumerPackets = consumerKeeper.GetPendingPackets(s.consumerCtx())
+	s.Require().Len(consumerPackets, 1, "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)
+	// go to next block to trigger SendPendingPackets
+	s.consumerChain.NextBlock()
+
+	// relay committed packets from consumer to provider, only second slash packet should be committed
+	relayAllCommittedPackets(s, s.consumerChain, s.path, ccv.ConsumerPortID, s.path.EndpointA.ChannelID, 1) // one slash
+
+	consumerPackets = consumerKeeper.GetPendingPackets(s.consumerCtx())
+	s.Require().Empty(consumerPackets, "pending data packets found")
 
 	// check that everything works
 	// - bond more tokens on provider to change validator powers
diff --git a/tests/integration/slashing.go b/tests/integration/slashing.go
index 2bc960fd03..1ce7d11db8 100644
--- a/tests/integration/slashing.go
+++ b/tests/integration/slashing.go
@@ -247,14 +247,34 @@ func (s *CCVTestSuite) TestSlashPacketAcknowledgement() {
 	s.SetupCCVChannel(s.path)
 	s.SetupTransferChannel()
 
-	packet := channeltypes.NewPacket([]byte{}, 1, ccv.ConsumerPortID, s.path.EndpointA.ChannelID,
+	// Mock a proper slash packet from consumer
+	spd := keepertestutil.GetNewSlashPacketData()
+
+	// We don't want truly randomized fields, infraction needs to be specified
+	if spd.Infraction == stakingtypes.Infraction_INFRACTION_UNSPECIFIED {
+		spd.Infraction = stakingtypes.Infraction_INFRACTION_DOUBLE_SIGN
+	}
+	cpd := ccv.NewConsumerPacketData(ccv.SlashPacket,
+		&ccv.ConsumerPacketData_SlashPacketData{
+			SlashPacketData: &spd,
+		},
+	)
+	packet := channeltypes.NewPacket(cpd.GetBytes(), // Consumer always sends v1 packet data
+		1, ccv.ConsumerPortID, s.path.EndpointA.ChannelID,
 		ccv.ProviderPortID, s.path.EndpointB.ChannelID, clienttypes.Height{}, 0)
 
-	ack := providerKeeper.OnRecvSlashPacket(s.providerCtx(), packet,
-		keepertestutil.GetNewSlashPacketData())
-	s.Require().NotNil(ack)
+	// Map infraction height on provider so validation passes and provider returns valid ack result
+	providerKeeper.SetValsetUpdateBlockHeight(s.providerCtx(), spd.ValsetUpdateId, 47923)
+
+	exportedAck := providerKeeper.OnRecvSlashPacket(s.providerCtx(), packet, spd)
+	s.Require().NotNil(exportedAck)
 
-	err := consumerKeeper.OnAcknowledgementPacket(s.consumerCtx(), packet, channeltypes.NewResultAcknowledgement(ack.Acknowledgement()))
+	// Unmarshal ack to struct that's compatible with consumer. IBC does this automatically
+	ack := channeltypes.Acknowledgement{}
+	err := channeltypes.SubModuleCdc.UnmarshalJSON(exportedAck.Acknowledgement(), &ack)
+	s.Require().NoError(err)
+
+	err = consumerKeeper.OnAcknowledgementPacket(s.consumerCtx(), packet, ack)
 	s.Require().NoError(err)
 
 	err = consumerKeeper.OnAcknowledgementPacket(s.consumerCtx(), packet, ccv.NewErrorAcknowledgementWithLog(s.consumerCtx(), fmt.Errorf("another error")))
@@ -492,9 +512,15 @@ func (suite *CCVTestSuite) TestValidatorDowntime() {
 	// clear queue, commit packets
 	suite.consumerApp.GetConsumerKeeper().SendPackets(ctx)
 
-	// check queue was cleared
+	// Check slash record is created
+	slashRecord, found := suite.consumerApp.GetConsumerKeeper().GetSlashRecord(suite.consumerCtx())
+	suite.Require().True(found, "slash record not found")
+	suite.Require().True(slashRecord.WaitingOnReply)
+	suite.Require().Equal(slashRecord.SendTime, suite.consumerCtx().BlockTime())
+
+	// check queue is not cleared, since no ack has been received from provider
 	pendingPackets = suite.consumerApp.GetConsumerKeeper().GetPendingPackets(ctx)
-	suite.Require().Empty(pendingPackets, "pending packets NOT empty")
+	suite.Require().Len(pendingPackets, 1, "pending packets len should be 1 is %d", len(pendingPackets))
 
 	// verify that the slash packet was sent
 	gotCommit := consumerIBCKeeper.ChannelKeeper.GetPacketCommitment(ctx, ccv.ConsumerPortID, channelID, seq)
@@ -579,9 +605,15 @@ func (suite *CCVTestSuite) TestValidatorDoubleSigning() {
 	// clear queue, commit packets
 	suite.consumerApp.GetConsumerKeeper().SendPackets(ctx)
 
-	// check queue was cleared
+	// Check slash record is created
+	slashRecord, found := suite.consumerApp.GetConsumerKeeper().GetSlashRecord(suite.consumerCtx())
+	suite.Require().True(found, "slash record not found")
+	suite.Require().True(slashRecord.WaitingOnReply)
+	suite.Require().Equal(slashRecord.SendTime, suite.consumerCtx().BlockTime())
+
+	// check queue is not cleared, since no ack has been received from provider
 	pendingPackets = suite.consumerApp.GetConsumerKeeper().GetPendingPackets(ctx)
-	suite.Require().Empty(pendingPackets, "pending packets NOT empty")
+	suite.Require().Len(pendingPackets, 1, "pending packets len should be 1 is %d", len(pendingPackets))
 
 	// check slash packet is sent
 	gotCommit := suite.consumerApp.GetIBCKeeper().ChannelKeeper.GetPacketCommitment(ctx, ccv.ConsumerPortID, channelID, seq)
@@ -644,10 +676,12 @@ func (suite *CCVTestSuite) TestQueueAndSendSlashPacket() {
 	// establish ccv channel by sending an empty VSC packet to consumer endpoint
 	suite.SendEmptyVSCPacket()
 
-	// check that each pending data packet is sent once
+	// check that each pending data packet is sent once, as long as the prev slash packet was relayed/acked.
+	// Note that consumer throttling blocks packet sending until a slash packet is successfully acked by the provider.
 	for i := 0; i < 12; i++ {
 		commit := consumerIBCKeeper.ChannelKeeper.GetPacketCommitment(ctx, ccv.ConsumerPortID, channelID, seq+uint64(i))
 		suite.Require().NotNil(commit)
+		relayAllCommittedPackets(suite, suite.consumerChain, suite.path, ccv.ConsumerPortID, channelID, 1)
 	}
 
 	// check that outstanding downtime flags
@@ -657,8 +691,8 @@ func (suite *CCVTestSuite) TestQueueAndSendSlashPacket() {
 		suite.Require().True(consumerKeeper.OutstandingDowntime(ctx, consAddr))
 	}
 
-	// send all pending packets - only slash packets should be queued in this test
-	consumerKeeper.SendPackets(ctx)
+	// SendPackets method should have already been called during
+	// endblockers in relayAllCommittedPackets above
 
 	// check that pending data packets got cleared
 	dataPackets = consumerKeeper.GetPendingPackets(ctx)
@@ -676,6 +710,10 @@ func (suite *CCVTestSuite) TestCISBeforeCCVEstablished() {
 	pendingPackets := consumerKeeper.GetPendingPackets(suite.consumerCtx())
 	suite.Require().Len(pendingPackets, 0)
 
+	// No slash record found (no slash sent)
+	_, found := consumerKeeper.GetSlashRecord(suite.consumerCtx())
+	suite.Require().False(found)
+
 	consumerKeeper.SlashWithInfractionReason(suite.consumerCtx(), []byte{0x01, 0x02, 0x3},
 		66, 4324, sdk.MustNewDecFromStr("0.05"), stakingtypes.Infraction_INFRACTION_DOWNTIME)
 
@@ -683,6 +721,10 @@ func (suite *CCVTestSuite) TestCISBeforeCCVEstablished() {
 	pendingPackets = consumerKeeper.GetPendingPackets(suite.consumerCtx())
 	suite.Require().Len(pendingPackets, 1)
 
+	// but slash packet is not yet sent
+	_, found = consumerKeeper.GetSlashRecord(suite.consumerCtx())
+	suite.Require().False(found)
+
 	// Pass 5 blocks, confirming the consumer doesn't panic
 	for i := 0; i < 5; i++ {
 		suite.consumerChain.NextBlock()
@@ -698,6 +740,8 @@ 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, 0)
+
+	// Slash record should now be found (slash sent)
+	_, found = consumerKeeper.GetSlashRecord(suite.consumerCtx())
+	suite.Require().True(found)
 }
diff --git a/tests/integration/throttle_retry.go b/tests/integration/throttle_retry.go
new file mode 100644
index 0000000000..ae15aac977
--- /dev/null
+++ b/tests/integration/throttle_retry.go
@@ -0,0 +1,150 @@
+package integration
+
+import (
+	channeltypes "github.com/cosmos/ibc-go/v7/modules/core/04-channel/types"
+
+	stakingtypes "github.com/cosmos/cosmos-sdk/x/staking/types"
+
+	provider "github.com/cosmos/interchain-security/v3/x/ccv/provider"
+	providertypes "github.com/cosmos/interchain-security/v3/x/ccv/provider/types"
+	ccvtypes "github.com/cosmos/interchain-security/v3/x/ccv/types"
+)
+
+// TestSlashRetries tests the throttling v2 retry logic. Without provider changes,
+// the consumer will queue up a slash packet, the provider will return a v1 result,
+// and the consumer will never need to retry.
+//
+// Once provider changes are made (slash packet queuing is removed), the consumer may retry packets
+// via new result acks from the provider.
+//
+// TODO: This test will need updating once provider changes are made.
+func (s *CCVTestSuite) TestSlashRetries() {
+	s.SetupAllCCVChannels()
+	s.setupValidatorPowers()
+
+	//
+	// Provider setup
+	//
+	providerKeeper := s.providerApp.GetProviderKeeper()
+	providerModule := provider.NewAppModule(&providerKeeper, s.providerApp.GetSubspace(providertypes.ModuleName))
+	// Initialize slash meter
+	providerKeeper.InitializeSlashMeter(s.providerCtx())
+	// Assert that we start out with no jailings
+	providerStakingKeeper := s.providerApp.GetTestStakingKeeper()
+	vals := providerStakingKeeper.GetAllValidators(s.providerCtx())
+	for _, val := range vals {
+		s.Require().False(val.IsJailed())
+	}
+	// Setup signing info for jailings
+	s.setDefaultValSigningInfo(*s.providerChain.Vals.Validators[1])
+
+	//
+	// Consumer setup
+	//
+	consumerKeeper := s.consumerApp.GetConsumerKeeper()
+	// Assert no slash record exists
+	_, found := consumerKeeper.GetSlashRecord(s.consumerCtx())
+	s.Require().False(found)
+
+	//
+	// Test section: See FSM explanation in throttle_retry.go
+	//
+
+	// Construct a mock slash packet from consumer
+	tmval1 := s.providerChain.Vals.Validators[1]
+	packet1 := s.constructSlashPacketFromConsumer(s.getFirstBundle(), *tmval1, stakingtypes.Infraction_INFRACTION_DOWNTIME, 1)
+
+	// Mock the sending of the packet on consumer
+	consumerKeeper.AppendPendingPacket(s.consumerCtx(), ccvtypes.SlashPacket,
+		&ccvtypes.ConsumerPacketData_SlashPacketData{
+			SlashPacketData: &ccvtypes.SlashPacketData{},
+		},
+	)
+	consumerKeeper.UpdateSlashRecordOnSend(s.consumerCtx())
+	slashRecord, found := consumerKeeper.GetSlashRecord(s.consumerCtx())
+	s.Require().True(found)
+	s.Require().True(slashRecord.WaitingOnReply)
+	s.Require().Len(consumerKeeper.GetPendingPackets(s.consumerCtx()), 1)
+
+	// Recv packet on provider and assert ack. Provider should return v1 result.
+	ack := providerModule.OnRecvPacket(s.providerCtx(), packet1, nil)
+	expectedv1Ack := channeltypes.NewResultAcknowledgement([]byte(ccvtypes.V1Result))
+	s.Require().Equal(expectedv1Ack.Acknowledgement(), ack.Acknowledgement())
+
+	// Couple blocks pass on provider for provider staking keeper to process jailing
+	s.providerChain.NextBlock()
+	s.providerChain.NextBlock()
+
+	// Default slash meter replenish fraction is 0.05, so packet should be handled on provider.
+	vals = s.providerApp.GetTestStakingKeeper().GetAllValidators(s.providerCtx())
+	s.Require().True(vals[1].IsJailed())
+	s.Require().Equal(int64(0),
+		s.providerApp.GetTestStakingKeeper().GetLastValidatorPower(s.providerCtx(), vals[1].GetOperator()))
+	s.Require().Equal(uint64(0), providerKeeper.GetThrottledPacketDataSize(s.providerCtx(),
+		s.getFirstBundle().Chain.ChainID))
+
+	// Now slash meter should be negative on provider
+	s.Require().True(s.providerApp.GetProviderKeeper().GetSlashMeter(s.providerCtx()).IsNegative())
+
+	// Apply ack back on consumer
+	ackForConsumer := expectedv1Ack
+	err := consumerKeeper.OnAcknowledgementPacket(s.consumerCtx(), packet1, ackForConsumer)
+	s.Require().NoError(err)
+
+	// Slash record should have been deleted, head of pending packets should have been popped
+	// Since provider has handled the packet
+	_, found = consumerKeeper.GetSlashRecord(s.consumerCtx())
+	s.Require().False(found)
+	s.Require().Empty(consumerKeeper.GetPendingPackets(s.consumerCtx()))
+
+	// pass two blocks on provider and consumer for good measure
+	s.providerChain.NextBlock()
+	s.providerChain.NextBlock()
+	s.consumerChain.NextBlock()
+	s.consumerChain.NextBlock()
+
+	// Construct and mock the sending of a second packet on consumer
+	tmval2 := s.providerChain.Vals.Validators[2]
+	packet2 := s.constructSlashPacketFromConsumer(s.getFirstBundle(), *tmval2, stakingtypes.Infraction_INFRACTION_DOWNTIME, 1)
+
+	consumerKeeper.AppendPendingPacket(s.consumerCtx(), ccvtypes.SlashPacket,
+		&ccvtypes.ConsumerPacketData_SlashPacketData{
+			SlashPacketData: &ccvtypes.SlashPacketData{},
+		},
+	)
+	consumerKeeper.UpdateSlashRecordOnSend(s.consumerCtx())
+	slashRecord, found = consumerKeeper.GetSlashRecord(s.consumerCtx())
+	s.Require().True(found)
+	s.Require().True(slashRecord.WaitingOnReply)
+	s.Require().Len(consumerKeeper.GetPendingPackets(s.consumerCtx()), 1)
+
+	// Recv 2nd slash packet on provider for different validator.
+	// Provider should return the same v1 result ack even tho the packet was queued.
+	ack = providerModule.OnRecvPacket(s.providerCtx(), packet2, nil)
+	expectedv1Ack = channeltypes.NewResultAcknowledgement([]byte(ccvtypes.V1Result))
+	s.Require().Equal(expectedv1Ack.Acknowledgement(), ack.Acknowledgement())
+
+	// Couple blocks pass on provider for staking keeper to process jailings
+	s.providerChain.NextBlock()
+	s.providerChain.NextBlock()
+
+	// Val shouldn't be jailed on provider. Slash packet was queued
+	s.Require().False(vals[2].IsJailed())
+	s.Require().Equal(int64(1000),
+		providerStakingKeeper.GetLastValidatorPower(s.providerCtx(), vals[2].GetOperator()))
+	s.Require().Equal(uint64(1), providerKeeper.GetThrottledPacketDataSize(s.providerCtx(),
+		s.getFirstBundle().Chain.ChainID))
+
+	// Apply ack on consumer
+	ackForConsumer = expectedv1Ack
+	err = consumerKeeper.OnAcknowledgementPacket(s.consumerCtx(), packet2, ackForConsumer)
+	s.Require().NoError(err)
+
+	// TODO: when provider changes are made, slashRecord.WaitingOnReply should have been updated to false on consumer. Slash Packet will still be in consumer's pending packets queue.
+
+	// Slash record should have been deleted, head of pending packets should have been popped
+	// Since provider has handled the packet
+	_, found = consumerKeeper.GetSlashRecord(s.consumerCtx())
+	s.Require().False(found)
+	s.Require().Empty(consumerKeeper.GetPendingPackets(s.consumerCtx()))
+}
diff --git a/testutil/integration/debug_test.go b/testutil/integration/debug_test.go
index 077f33cde3..828c9b6810 100644
--- a/testutil/integration/debug_test.go
+++ b/testutil/integration/debug_test.go
@@ -252,3 +252,11 @@ func TestQueueAndSendVSCMaturedPackets(t *testing.T) {
 func TestRecycleTransferChannel(t *testing.T) {
 	runCCVTestByName(t, "TestRecycleTransferChannel")
 }
+
+//
+// Throttle retry tests
+//
+
+func TestSlashRetries(t *testing.T) {
+	runCCVTestByName(t, "TestSlashRetries")
+}
diff --git a/x/ccv/consumer/keeper/keeper.go b/x/ccv/consumer/keeper/keeper.go
index e8b1cb793e..94d5c790fd 100644
--- a/x/ccv/consumer/keeper/keeper.go
+++ b/x/ccv/consumer/keeper/keeper.go
@@ -606,6 +606,17 @@ func (k Keeper) getAndIncrementPendingPacketsIdx(ctx sdk.Context) (toReturn uint
 	return toReturn
 }
 
+// DeleteHeadOfPendingPackets deletes the head of the pending packets queue.
+func (k Keeper) DeleteHeadOfPendingPackets(ctx sdk.Context) {
+	store := ctx.KVStore(k.storeKey)
+	iterator := sdk.KVStorePrefixIterator(store, []byte{types.PendingDataPacketsBytePrefix})
+	defer iterator.Close()
+	if !iterator.Valid() {
+		return
+	}
+	store.Delete(iterator.Key())
+}
+
 // GetPendingPackets returns ALL the pending CCV packets from the store without indexes.
 func (k Keeper) GetPendingPackets(ctx sdk.Context) []ccv.ConsumerPacketData {
 	ppWithIndexes := k.GetAllPendingPacketsWithIdx(ctx)
diff --git a/x/ccv/consumer/keeper/keeper_test.go b/x/ccv/consumer/keeper/keeper_test.go
index 269c60d9c5..06fdeae082 100644
--- a/x/ccv/consumer/keeper/keeper_test.go
+++ b/x/ccv/consumer/keeper/keeper_test.go
@@ -579,3 +579,30 @@ func TestPrevStandaloneChainFlag(t *testing.T) {
 	ck.MarkAsPrevStandaloneChain(ctx)
 	require.True(t, ck.IsPrevStandaloneChain(ctx))
 }
+
+func TestDeleteHeadOfPendingPackets(t *testing.T) {
+	consumerKeeper, ctx, ctrl, _ := testkeeper.GetConsumerKeeperAndCtx(t, testkeeper.NewInMemKeeperParams(t))
+	defer ctrl.Finish()
+
+	// append some pending packets
+	consumerKeeper.AppendPendingPacket(ctx, ccv.VscMaturedPacket, &ccv.ConsumerPacketData_VscMaturedPacketData{})
+	consumerKeeper.AppendPendingPacket(ctx, ccv.SlashPacket, &ccv.ConsumerPacketData_SlashPacketData{})
+	consumerKeeper.AppendPendingPacket(ctx, ccv.VscMaturedPacket, &ccv.ConsumerPacketData_VscMaturedPacketData{})
+
+	// Check there's 3 pending packets, vsc matured at head
+	pp := consumerKeeper.GetPendingPackets(ctx)
+	require.Len(t, pp, 3)
+	require.Equal(t, pp[0].Type, ccv.VscMaturedPacket)
+
+	// Delete the head, confirm slash packet is now at head
+	consumerKeeper.DeleteHeadOfPendingPackets(ctx)
+	pp = consumerKeeper.GetPendingPackets(ctx)
+	require.Len(t, pp, 2)
+	require.Equal(t, pp[0].Type, ccv.SlashPacket)
+
+	// Delete the head, confirm vsc matured packet is now at head
+	consumerKeeper.DeleteHeadOfPendingPackets(ctx)
+	pp = consumerKeeper.GetPendingPackets(ctx)
+	require.Len(t, pp, 1)
+	require.Equal(t, pp[0].Type, ccv.VscMaturedPacket)
+}
diff --git a/x/ccv/consumer/keeper/relay.go b/x/ccv/consumer/keeper/relay.go
index 376110f26d..060aadff20 100644
--- a/x/ccv/consumer/keeper/relay.go
+++ b/x/ccv/consumer/keeper/relay.go
@@ -188,8 +188,11 @@ func (k Keeper) SendPackets(ctx sdk.Context) {
 	pending := k.GetAllPendingPacketsWithIdx(ctx)
 	idxsForDeletion := []uint64{}
 	for _, p := range pending {
+		if !k.PacketSendingPermitted(ctx) {
+			break
+		}
 
-		// send packet over IBC
+		// Send packet over IBC
 		err := ccv.SendIBCPacket(
 			ctx,
 			k.scopedKeeper,
@@ -213,6 +216,16 @@ func (k Keeper) SendPackets(ctx sdk.Context) {
 			k.Logger(ctx).Error("cannot send IBC packet; leaving packet data stored:", "type", p.Type.String(), "err", err.Error())
 			break
 		}
+		// If the packet that was just sent was a Slash packet, set the waiting on slash reply flag.
+		// This flag will be toggled false again when consumer hears back from provider. See OnAcknowledgementPacket below.
+		if p.Type == ccv.SlashPacket {
+			k.UpdateSlashRecordOnSend(ctx)
+			// Break so slash stays at head of queue.
+			// This blocks the sending of any other packet until the leading slash packet is handled.
+			// Also see OnAcknowledgementPacket below which will eventually delete the leading slash packet.
+			break
+		}
+		// Otherwise the vsc matured will be deleted
 		idxsForDeletion = append(idxsForDeletion, p.Idx)
 	}
 	// Delete pending packets that were successfully sent and did not return an error from SendIBCPacket
@@ -223,6 +236,40 @@ func (k Keeper) SendPackets(ctx sdk.Context) {
 // in conjunction with the ibc module's execution of "acknowledgePacket",
 // according to https://github.com/cosmos/ibc/tree/main/spec/core/ics-004-channel-and-packet-semantics#processing-acknowledgements
 func (k Keeper) OnAcknowledgementPacket(ctx sdk.Context, packet channeltypes.Packet, ack channeltypes.Acknowledgement) error {
+	if res := ack.GetResult(); res != nil {
+		if len(res) != 1 {
+			return fmt.Errorf("acknowledgement result length must be 1, got %d", len(res))
+		}
+
+		// Unmarshal into V1 consumer packet data type. We trust data is formed correctly
+		// as it was originally marshalled by this module, and consumers must trust the provider
+		// did not tamper with the data. Note ConsumerPacketData.GetBytes() always JSON marshals to the
+		// ConsumerPacketDataV1 type which is sent over the wire.
+		var consumerPacket ccv.ConsumerPacketDataV1
+		ccv.ModuleCdc.MustUnmarshalJSON(packet.GetData(), &consumerPacket)
+		// If this ack is regarding a provider handling a vsc matured packet, there's nothing to do.
+		// As vsc matured packets are popped from the consumer pending packets queue on send.
+		if consumerPacket.Type == ccv.VscMaturedPacket {
+			return nil
+		}
+
+		// Otherwise we handle the result of the slash packet acknowledgement.
+		switch res[0] {
+		// We treat a v1 result as the provider successfully queuing the slash packet w/o need for retry.
+		case ccv.V1Result[0]:
+			k.ClearSlashRecord(ctx)           // Clears slash record state, unblocks sending of pending packets.
+			k.DeleteHeadOfPendingPackets(ctx) // Remove slash from head of queue. It's been handled.
+		case ccv.SlashPacketHandledResult[0]:
+			k.ClearSlashRecord(ctx)           // Clears slash record state, unblocks sending of pending packets.
+			k.DeleteHeadOfPendingPackets(ctx) // Remove slash from head of queue. It's been handled.
+		case ccv.SlashPacketBouncedResult[0]:
+			k.UpdateSlashRecordOnBounce(ctx)
+			// Note slash is still at head of queue and will now be retried after appropriate delay period.
+		default:
+			return fmt.Errorf("unrecognized acknowledgement result: %c", res[0])
+		}
+	}
+
 	if err := ack.GetError(); err != "" {
 		// Reasons for ErrorAcknowledgment
 		//  - packet data could not be successfully decoded
diff --git a/x/ccv/consumer/keeper/relay_test.go b/x/ccv/consumer/keeper/relay_test.go
index 6dbebe273f..2fe90608c1 100644
--- a/x/ccv/consumer/keeper/relay_test.go
+++ b/x/ccv/consumer/keeper/relay_test.go
@@ -14,6 +14,7 @@ import (
 
 	cryptocodec "github.com/cosmos/cosmos-sdk/crypto/codec"
 	"github.com/cosmos/cosmos-sdk/crypto/keys/ed25519"
+	sdk "github.com/cosmos/cosmos-sdk/types"
 	capabilitytypes "github.com/cosmos/cosmos-sdk/x/capability/types"
 	stakingtypes "github.com/cosmos/cosmos-sdk/x/staking/types"
 
@@ -22,6 +23,7 @@ import (
 
 	"github.com/cosmos/interchain-security/v3/testutil/crypto"
 	testkeeper "github.com/cosmos/interchain-security/v3/testutil/keeper"
+	consumerkeeper "github.com/cosmos/interchain-security/v3/x/ccv/consumer/keeper"
 	consumertypes "github.com/cosmos/interchain-security/v3/x/ccv/consumer/types"
 	"github.com/cosmos/interchain-security/v3/x/ccv/types"
 )
@@ -210,10 +212,105 @@ func TestOnRecvVSCPacketDuplicateUpdates(t *testing.T) {
 	require.Equal(t, valUpdates[1], gotPendingChanges.ValidatorUpdates[0]) // Only latest update should be kept
 }
 
-// TestOnAcknowledgementPacket tests application logic for acknowledgments of sent VSCMatured and Slash packets
+// TestSendPackets tests the SendPackets method failing
+func TestSendPacketsFailure(t *testing.T) {
+	// Keeper setup
+	consumerKeeper, ctx, ctrl, mocks := testkeeper.GetConsumerKeeperAndCtx(t, testkeeper.NewInMemKeeperParams(t))
+	defer ctrl.Finish()
+	consumerKeeper.SetProviderChannel(ctx, "consumerCCVChannelID")
+	consumerKeeper.SetParams(ctx, consumertypes.DefaultParams())
+
+	// Set some pending packets
+	consumerKeeper.AppendPendingPacket(ctx, types.VscMaturedPacket, &types.ConsumerPacketData_VscMaturedPacketData{})
+	consumerKeeper.AppendPendingPacket(ctx, types.SlashPacket, &types.ConsumerPacketData_SlashPacketData{})
+	consumerKeeper.AppendPendingPacket(ctx, types.VscMaturedPacket, &types.ConsumerPacketData_VscMaturedPacketData{})
+
+	// Mock the channel keeper to return an error
+	gomock.InOrder(
+		mocks.MockChannelKeeper.EXPECT().GetChannel(ctx, types.ConsumerPortID,
+			"consumerCCVChannelID").Return(channeltypes.Channel{}, false).Times(1),
+	)
+
+	// No panic should occur, pending packets should not be cleared
+	consumerKeeper.SendPackets(ctx)
+	require.Equal(t, 3, len(consumerKeeper.GetPendingPackets(ctx)))
+}
+
+func TestSendPackets(t *testing.T) {
+	// Keeper setup
+	consumerKeeper, ctx, ctrl, mocks := testkeeper.GetConsumerKeeperAndCtx(t, testkeeper.NewInMemKeeperParams(t))
+	consumerKeeper.SetProviderChannel(ctx, "consumerCCVChannelID")
+	consumerKeeper.SetParams(ctx, consumertypes.DefaultParams())
+
+	// No slash record should exist
+	_, found := consumerKeeper.GetSlashRecord(ctx)
+	require.False(t, found)
+	require.True(t, consumerKeeper.PacketSendingPermitted(ctx))
+
+	// Queue up two vsc matured, followed by slash, followed by vsc matured
+	consumerKeeper.AppendPendingPacket(ctx, types.VscMaturedPacket, &types.ConsumerPacketData_VscMaturedPacketData{
+		VscMaturedPacketData: &types.VSCMaturedPacketData{
+			ValsetUpdateId: 77,
+		},
+	})
+	consumerKeeper.AppendPendingPacket(ctx, types.VscMaturedPacket, &types.ConsumerPacketData_VscMaturedPacketData{
+		VscMaturedPacketData: &types.VSCMaturedPacketData{
+			ValsetUpdateId: 90,
+		},
+	})
+	consumerKeeper.AppendPendingPacket(ctx, types.SlashPacket, &types.ConsumerPacketData_SlashPacketData{
+		SlashPacketData: &types.SlashPacketData{
+			Validator:      abci.Validator{},
+			ValsetUpdateId: 88,
+			Infraction:     stakingtypes.Infraction_INFRACTION_DOWNTIME,
+		},
+	})
+	consumerKeeper.AppendPendingPacket(ctx, types.VscMaturedPacket, &types.ConsumerPacketData_VscMaturedPacketData{
+		VscMaturedPacketData: &types.VSCMaturedPacketData{
+			ValsetUpdateId: 99,
+		},
+	})
+
+	// First two vsc matured and slash should be sent, 3 total
+	gomock.InAnyOrder(
+		testkeeper.GetMocksForSendIBCPacket(ctx, mocks, "consumerCCVChannelID", 3),
+	)
+	consumerKeeper.SendPackets(ctx)
+	ctrl.Finish()
+
+	// First two packets should be deleted, slash should be at head of queue
+	pendingPackets := consumerKeeper.GetPendingPackets(ctx)
+	require.Equal(t, 2, len(pendingPackets))
+	require.Equal(t, types.SlashPacket, pendingPackets[0].Type)
+	require.Equal(t, types.VscMaturedPacket, pendingPackets[1].Type)
+
+	// Packet sending not permitted
+	require.False(t, consumerKeeper.PacketSendingPermitted(ctx))
+
+	// Now delete slash record as would be done by a recv SlashPacketHandledResult
+	// then confirm last vsc matured is sent
+	consumerKeeper.ClearSlashRecord(ctx)
+	consumerKeeper.DeleteHeadOfPendingPackets(ctx)
+
+	// Packet sending permitted
+	require.True(t, consumerKeeper.PacketSendingPermitted(ctx))
+
+	gomock.InAnyOrder(
+		testkeeper.GetMocksForSendIBCPacket(ctx, mocks, "consumerCCVChannelID", 1),
+	)
+
+	consumerKeeper.SendPackets(ctx)
+	ctrl.Finish()
+
+	// No packets should be left
+	pendingPackets = consumerKeeper.GetPendingPackets(ctx)
+	require.Equal(t, 0, len(pendingPackets))
+}
+
+// TestOnAcknowledgementPacketError tests application logic for ERROR acknowledgments of sent VSCMatured and Slash packets
 // in conjunction with the ibc module's execution of "acknowledgePacket",
 // according to https://github.com/cosmos/ibc/tree/main/spec/core/ics-004-channel-and-packet-semantics#processing-acknowledgements
-func TestOnAcknowledgementPacket(t *testing.T) {
+func TestOnAcknowledgementPacketError(t *testing.T) {
 	// Channel ID to some dest chain that's not the established provider
 	channelIDToDestChain := "channelIDToDestChain"
 
@@ -250,12 +347,6 @@ func TestOnAcknowledgementPacket(t *testing.T) {
 		uint64(time.Now().Add(60*time.Second).UnixNano()),
 	)
 
-	ack := channeltypes.NewResultAcknowledgement([]byte{1})
-
-	// expect no error returned from OnAcknowledgementPacket, no input error with ack
-	err := consumerKeeper.OnAcknowledgementPacket(ctx, packet, ack)
-	require.Nil(t, err)
-
 	// Still expect no error returned from OnAcknowledgementPacket,
 	// but the input error ack will be handled with appropriate ChanCloseInit calls
 	dummyCap := &capabilitytypes.Capability{}
@@ -279,33 +370,91 @@ func TestOnAcknowledgementPacket(t *testing.T) {
 		).Return(nil).Times(1),
 	)
 
-	ack = types.NewErrorAcknowledgementWithLog(ctx, fmt.Errorf("error"))
-	err = consumerKeeper.OnAcknowledgementPacket(ctx, packet, ack)
+	ack := types.NewErrorAcknowledgementWithLog(ctx, fmt.Errorf("error"))
+	err := consumerKeeper.OnAcknowledgementPacket(ctx, packet, ack)
 	require.Nil(t, err)
 }
 
-// TestSendPackets tests the SendPackets method failing
-func TestSendPacketsFailure(t *testing.T) {
-	// Keeper setup
-	consumerKeeper, ctx, ctrl, mocks := testkeeper.GetConsumerKeeperAndCtx(t, testkeeper.NewInMemKeeperParams(t))
+// TestOnAcknowledgementPacketResult tests application logic for RESULT acknowledgments of sent VSCMatured and Slash packets
+// in conjunction with the ibc module's execution of "acknowledgePacket",
+func TestOnAcknowledgementPacketResult(t *testing.T) {
+	// Setup
+	consumerKeeper, ctx, ctrl, _ := testkeeper.GetConsumerKeeperAndCtx(t, testkeeper.NewInMemKeeperParams(t))
 	defer ctrl.Finish()
-	consumerKeeper.SetProviderChannel(ctx, "consumerCCVChannelID")
-	consumerKeeper.SetParams(ctx, consumertypes.DefaultParams())
 
-	// Set some pending packets
-	consumerKeeper.AppendPendingPacket(ctx, types.VscMaturedPacket, &types.ConsumerPacketData_VscMaturedPacketData{})
-	consumerKeeper.AppendPendingPacket(ctx, types.SlashPacket, &types.ConsumerPacketData_SlashPacketData{})
-	consumerKeeper.AppendPendingPacket(ctx, types.VscMaturedPacket, &types.ConsumerPacketData_VscMaturedPacketData{})
+	setupSlashBeforeVscMatured(ctx, &consumerKeeper)
 
-	// Mock the channel keeper to return an error
-	gomock.InOrder(
-		mocks.MockChannelKeeper.EXPECT().GetChannel(ctx, types.ConsumerPortID,
-			"consumerCCVChannelID").Return(channeltypes.Channel{}, false).Times(1),
-	)
+	// Slash record found, 2 pending packets, slash is at head of queue
+	_, found := consumerKeeper.GetSlashRecord(ctx)
+	require.True(t, found)
+	pendingPackets := consumerKeeper.GetPendingPackets(ctx)
+	require.Len(t, pendingPackets, 2)
+	require.Equal(t, types.SlashPacket, pendingPackets[0].Type)
 
-	// No panic should occur, pending packets should not be cleared
-	consumerKeeper.SendPackets(ctx)
-	require.Equal(t, 3, len(consumerKeeper.GetPendingPackets(ctx)))
+	// v1 result should delete slash record and head of pending packets. Vsc matured remains
+	ack := channeltypes.NewResultAcknowledgement(types.V1Result)
+	packet := channeltypes.Packet{Data: pendingPackets[0].GetBytes()}
+	err := consumerKeeper.OnAcknowledgementPacket(ctx, packet, ack)
+	require.Nil(t, err)
+	_, found = consumerKeeper.GetSlashRecord(ctx)
+	require.False(t, found)
+	require.Len(t, consumerKeeper.GetPendingPackets(ctx), 1)
+	require.Equal(t, types.VscMaturedPacket, consumerKeeper.GetPendingPackets(ctx)[0].Type)
+
+	// refresh state
+	setupSlashBeforeVscMatured(ctx, &consumerKeeper)
+	pendingPackets = consumerKeeper.GetPendingPackets(ctx)
+	packet = channeltypes.Packet{Data: pendingPackets[0].GetBytes()}
+
+	// Slash packet handled result should delete slash record and head of pending packets
+	ack = channeltypes.NewResultAcknowledgement(types.SlashPacketHandledResult)
+	err = consumerKeeper.OnAcknowledgementPacket(ctx, packet, ack)
+	require.Nil(t, err)
+	_, found = consumerKeeper.GetSlashRecord(ctx)
+	require.False(t, found)
+	require.Len(t, consumerKeeper.GetPendingPackets(ctx), 1)
+	require.Equal(t, types.VscMaturedPacket, consumerKeeper.GetPendingPackets(ctx)[0].Type)
+
+	// refresh state
+	setupSlashBeforeVscMatured(ctx, &consumerKeeper)
+	pendingPackets = consumerKeeper.GetPendingPackets(ctx)
+	packet = channeltypes.Packet{Data: pendingPackets[0].GetBytes()}
+
+	slashRecordBefore, found := consumerKeeper.GetSlashRecord(ctx)
+	require.True(t, found)
+	require.True(t, slashRecordBefore.WaitingOnReply)
+
+	// Slash packet bounced result should update slash record
+	ack = channeltypes.NewResultAcknowledgement(types.SlashPacketBouncedResult)
+	err = consumerKeeper.OnAcknowledgementPacket(ctx, packet, ack)
+	require.Nil(t, err)
+	slashRecordAfter, found := consumerKeeper.GetSlashRecord(ctx)
+	require.True(t, found)
+	require.False(t, slashRecordAfter.WaitingOnReply) // waiting on reply toggled false
+	require.Equal(t, slashRecordAfter.SendTime.UnixNano(),
+		slashRecordBefore.SendTime.UnixNano()) // send time NOT updated. Bounce result shouldn't affect that
+}
+
+func setupSlashBeforeVscMatured(ctx sdk.Context, k *consumerkeeper.Keeper) {
+	// clear old state
+	k.ClearSlashRecord(ctx)
+	k.DeleteAllPendingDataPackets(ctx)
+
+	// Set some related state to test against
+	k.SetSlashRecord(ctx, consumertypes.SlashRecord{WaitingOnReply: true, SendTime: time.Now()})
+	// Slash packet before VSCMatured packet
+	k.AppendPendingPacket(ctx, types.SlashPacket, &types.ConsumerPacketData_SlashPacketData{ // Slash appears first
+		SlashPacketData: &types.SlashPacketData{
+			Validator:      abci.Validator{},
+			ValsetUpdateId: 88,
+			Infraction:     stakingtypes.Infraction_INFRACTION_DOWNTIME,
+		},
+	})
+	k.AppendPendingPacket(ctx, types.VscMaturedPacket, &types.ConsumerPacketData_VscMaturedPacketData{
+		VscMaturedPacketData: &types.VSCMaturedPacketData{
+			ValsetUpdateId: 90,
+		},
+	})
 }
 
 // Regression test for https://github.com/cosmos/interchain-security/issues/1145
@@ -316,7 +465,12 @@ func TestSendPacketsDeletion(t *testing.T) {
 	consumerKeeper.SetProviderChannel(ctx, "consumerCCVChannelID")
 	consumerKeeper.SetParams(ctx, consumertypes.DefaultParams())
 
-	// Queue two pending packets
+	// Queue two pending packets, vsc matured first
+	consumerKeeper.AppendPendingPacket(ctx, types.VscMaturedPacket, &types.ConsumerPacketData_VscMaturedPacketData{
+		VscMaturedPacketData: &types.VSCMaturedPacketData{
+			ValsetUpdateId: 90,
+		},
+	})
 	consumerKeeper.AppendPendingPacket(ctx, types.SlashPacket, &types.ConsumerPacketData_SlashPacketData{ // Slash appears first
 		SlashPacketData: &types.SlashPacketData{
 			Validator:      abci.Validator{},
@@ -324,15 +478,10 @@ func TestSendPacketsDeletion(t *testing.T) {
 			Infraction:     stakingtypes.Infraction_INFRACTION_DOWNTIME,
 		},
 	})
-	consumerKeeper.AppendPendingPacket(ctx, types.VscMaturedPacket, &types.ConsumerPacketData_VscMaturedPacketData{
-		VscMaturedPacketData: &types.VSCMaturedPacketData{
-			ValsetUpdateId: 90,
-		},
-	})
 
-	// Get mocks for a successful SendPacket call that does NOT return an error
+	// Get mocks for the (first) successful SendPacket call that does NOT return an error
 	expectations := testkeeper.GetMocksForSendIBCPacket(ctx, mocks, "consumerCCVChannelID", 1)
-	// Append mocks for a failed SendPacket call, which returns an error
+	// Append mocks for the (second) failed SendPacket call, which returns an error
 	expectations = append(expectations, mocks.MockChannelKeeper.EXPECT().GetChannel(ctx, types.ConsumerPortID,
 		"consumerCCVChannelID").Return(channeltypes.Channel{}, false).Times(1))
 	gomock.InOrder(expectations...)
@@ -341,5 +490,7 @@ func TestSendPacketsDeletion(t *testing.T) {
 
 	// Expect the first successfully sent packet to be popped from queue
 	require.Equal(t, 1, len(consumerKeeper.GetPendingPackets(ctx)))
-	require.Equal(t, types.VscMaturedPacket, consumerKeeper.GetPendingPackets(ctx)[0].Type)
+
+	// Expect the slash packet to remain
+	require.Equal(t, types.SlashPacket, consumerKeeper.GetPendingPackets(ctx)[0].Type)
 }
diff --git a/x/ccv/consumer/keeper/throttle_retry.go b/x/ccv/consumer/keeper/throttle_retry.go
new file mode 100644
index 0000000000..4c4585cb1d
--- /dev/null
+++ b/x/ccv/consumer/keeper/throttle_retry.go
@@ -0,0 +1,112 @@
+package keeper
+
+import (
+	"fmt"
+	"time"
+
+	sdktypes "github.com/cosmos/cosmos-sdk/types"
+
+	consumertypes "github.com/cosmos/interchain-security/v3/x/ccv/consumer/types"
+)
+
+//
+// Throttling with retries follows a finite-state machine design:
+//
+// 2 states: "No Slash" and "Standby".
+// Initial State: "No Slash"
+// Transition Event: ("No Slash", Slash packet sent) => ("Standby")
+// Transition Event: ("Standby", V1Result ack received) => ("No Slash")
+// Transition Event: ("Standby", Slash packet successfully handled) => ("No Slash")
+// Internal Transition Event: ("Standby", Slash packet bounced) => ("Standby", with SlashRecord.WaitingOnReply = false)
+// Transition Event: ("Standby", Retry sent) => ("Standby", new cycle)
+//
+// Description in words:
+//
+// 1. "No slash": If no slash record exists, the consumer is permitted to send packets from the pending packets queue.
+// The consumer starts in this state from genesis.
+//
+// 2. On the event that a slash packet is obtained from the head of the pending packets queue and sent,
+// a consumer transitions from "No Slash" to "Standby". A slash record is created upon entry to this state,
+// and the consumer is now restricted from sending anymore packets.
+//
+// The slash packet remains at the head of the pending packets queue within the "Standby" state.
+//
+// - If the consumer receives a V1Result ack from the provider,
+// OR if the consumer receives an ack from the provider that the slash packet was successfully handled,
+// the consumer transitions from "Standby" to "No Slash".
+// The slash record is cleared upon this transition, and the slash packet is popped from the pending packets queue.
+//
+// - Else if the consumer receives an ack from the provider that the slash packet was bounced (not handled),
+// then SlashRecord.WaitingOnReply is set false, and the consumer retries sending the slash packet after a delay period.
+//
+// Once a retry is sent, the consumer enters a new cycle of the "Standby" state and the process repeats.
+//
+// This design is implemented below, and in relay.go under SendPackets() and OnAcknowledgementPacket().
+//
+
+// Retry delay period could be implemented as a param, but 1 hour is reasonable
+const RetryDelayPeriod = time.Hour
+
+// PacketSendingPermitted returns whether the consumer is allowed to send packets
+// from the pending packets queue.
+func (k Keeper) PacketSendingPermitted(ctx sdktypes.Context) bool {
+	record, found := k.GetSlashRecord(ctx)
+	if !found {
+		// no slash record exists, send is permitted
+		return true
+	}
+	if record.WaitingOnReply {
+		// We are waiting on a reply from provider, block sending
+		return false
+	}
+	// If retry delay period has elapsed, we can send again
+	return ctx.BlockTime().After(record.SendTime.Add(RetryDelayPeriod))
+}
+
+func (k Keeper) UpdateSlashRecordOnSend(ctx sdktypes.Context) {
+	record := consumertypes.NewSlashRecord(
+		ctx.BlockTime(), // sendTime
+		true,            // waitingOnReply
+	)
+	// We don't mind overwriting here, since this is either a retry or the first time we send a slash
+	k.SetSlashRecord(ctx, record)
+}
+
+func (k Keeper) UpdateSlashRecordOnBounce(ctx sdktypes.Context) {
+	record, found := k.GetSlashRecord(ctx)
+	if !found {
+		// This should never happen
+		panic("could not find slash record, but reply was received from provider")
+	}
+	record.WaitingOnReply = false
+	k.SetSlashRecord(ctx, record)
+}
+
+func (k Keeper) GetSlashRecord(ctx sdktypes.Context) (record consumertypes.SlashRecord, found bool) {
+	store := ctx.KVStore(k.storeKey)
+	bz := store.Get(consumertypes.SlashRecordKey())
+	if bz == nil {
+		return record, false
+	}
+	err := record.Unmarshal(bz)
+	if err != nil {
+		// This should never happen
+		panic(fmt.Sprintf("could not unmarshal slash record: %v", err))
+	}
+	return record, true
+}
+
+func (k Keeper) SetSlashRecord(ctx sdktypes.Context, record consumertypes.SlashRecord) {
+	store := ctx.KVStore(k.storeKey)
+	bz, err := record.Marshal()
+	if err != nil {
+		// This should never happen
+		panic(fmt.Sprintf("could not marshal slash record: %v", err))
+	}
+	store.Set(consumertypes.SlashRecordKey(), bz)
+}
+
+func (k Keeper) ClearSlashRecord(ctx sdktypes.Context) {
+	store := ctx.KVStore(k.storeKey)
+	store.Delete(consumertypes.SlashRecordKey())
+}
diff --git a/x/ccv/consumer/keeper/throttle_retry_test.go b/x/ccv/consumer/keeper/throttle_retry_test.go
new file mode 100644
index 0000000000..cc14ce3cdd
--- /dev/null
+++ b/x/ccv/consumer/keeper/throttle_retry_test.go
@@ -0,0 +1,91 @@
+package keeper_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	testutil "github.com/cosmos/interchain-security/v3/testutil/keeper"
+	consumerkeeper "github.com/cosmos/interchain-security/v3/x/ccv/consumer/keeper"
+	consumertypes "github.com/cosmos/interchain-security/v3/x/ccv/consumer/types"
+)
+
+func TestPacketSendingPermitted(t *testing.T) {
+	consumerKeeper, ctx, ctrl, _ := testutil.GetConsumerKeeperAndCtx(t, testutil.NewInMemKeeperParams(t))
+	defer ctrl.Finish()
+
+	ctx = ctx.WithBlockTime(time.Now())
+
+	// No slash record exists, send is permitted
+	slashRecord, found := consumerKeeper.GetSlashRecord(ctx)
+	require.False(t, found)
+	require.Zero(t, slashRecord)
+	require.True(t, consumerKeeper.PacketSendingPermitted(ctx))
+
+	// Update slash record on sending of slash packet
+	consumerKeeper.UpdateSlashRecordOnSend(ctx)
+	slashRecord, found = consumerKeeper.GetSlashRecord(ctx)
+	require.True(t, found)
+	require.True(t, slashRecord.WaitingOnReply)
+
+	// Packet sending not permitted since we're waiting on a reply from provider
+	require.False(t, consumerKeeper.PacketSendingPermitted(ctx))
+
+	// Call update that happens when provider bounces slash packet
+	consumerKeeper.UpdateSlashRecordOnBounce(ctx)
+	slashRecord, found = consumerKeeper.GetSlashRecord(ctx)
+	require.True(t, found)
+	require.False(t, slashRecord.WaitingOnReply)
+
+	// Packet sending still not permitted since retry delay period has not elapsed
+	require.False(t, consumerKeeper.PacketSendingPermitted(ctx))
+
+	// Elapse retry delay period
+	ctx = ctx.WithBlockTime(ctx.BlockTime().Add(2 * consumerkeeper.RetryDelayPeriod))
+
+	// Now packet sending is permitted again
+	require.True(t, consumerKeeper.PacketSendingPermitted(ctx))
+}
+
+func TestThrottleRetryCRUD(t *testing.T) {
+	consumerKeeper, ctx, ctrl, _ := testutil.GetConsumerKeeperAndCtx(t, testutil.NewInMemKeeperParams(t))
+	defer ctrl.Finish()
+
+	slashRecord, found := consumerKeeper.GetSlashRecord(ctx)
+	require.False(t, found)
+	require.Zero(t, slashRecord)
+
+	consumerKeeper.SetSlashRecord(ctx, consumertypes.SlashRecord{
+		WaitingOnReply: true,
+		SendTime:       ctx.BlockTime(),
+	})
+
+	slashRecord, found = consumerKeeper.GetSlashRecord(ctx)
+	require.True(t, found)
+	require.True(t, slashRecord.WaitingOnReply)
+	require.Equal(t, ctx.BlockTime(), slashRecord.SendTime)
+
+	// UpdateSlashRecordOnBounce should set WaitingOnReply to false, and leave SendTime unchanged
+	oldBlocktime := ctx.BlockTime()
+	ctx = ctx.WithBlockTime(ctx.BlockTime().Add(time.Hour))
+	consumerKeeper.UpdateSlashRecordOnBounce(ctx)
+	slashRecord, found = consumerKeeper.GetSlashRecord(ctx)
+	require.True(t, found)
+	require.False(t, slashRecord.WaitingOnReply)
+	require.Equal(t, oldBlocktime, slashRecord.SendTime) // Old SendTime expected
+
+	// UpdateSlashRecordOnSend should replace slash record with WaitingOnReply set to true, and new SendTime
+	ctx = ctx.WithBlockTime(ctx.BlockTime().Add(time.Hour))
+	consumerKeeper.UpdateSlashRecordOnSend(ctx)
+	slashRecord, found = consumerKeeper.GetSlashRecord(ctx)
+	require.True(t, found)
+	require.True(t, slashRecord.WaitingOnReply)
+	require.Equal(t, ctx.BlockTime(), slashRecord.SendTime)               // New SendTime expected
+	require.Equal(t, oldBlocktime.Add(2*time.Hour), slashRecord.SendTime) // Sanity check
+
+	consumerKeeper.ClearSlashRecord(ctx)
+	slashRecord, found = consumerKeeper.GetSlashRecord(ctx)
+	require.False(t, found)
+	require.Zero(t, slashRecord)
+}
diff --git a/x/ccv/consumer/types/consumer.pb.go b/x/ccv/consumer/types/consumer.pb.go
index 90d5d6e12b..b16b561b7b 100644
--- a/x/ccv/consumer/types/consumer.pb.go
+++ b/x/ccv/consumer/types/consumer.pb.go
@@ -354,11 +354,66 @@ func (m *MaturingVSCPacket) GetMaturityTime() time.Time {
 	return time.Time{}
 }
 
+// A record storing the state of a slash packet sent to the provider chain
+// which may bounce back and forth until handled by the provider.
+type SlashRecord struct {
+	WaitingOnReply bool      `protobuf:"varint,1,opt,name=waiting_on_reply,json=waitingOnReply,proto3" json:"waiting_on_reply,omitempty"`
+	SendTime       time.Time `protobuf:"bytes,2,opt,name=send_time,json=sendTime,proto3,stdtime" json:"send_time"`
+}
+
+func (m *SlashRecord) Reset()         { *m = SlashRecord{} }
+func (m *SlashRecord) String() string { return proto.CompactTextString(m) }
+func (*SlashRecord) ProtoMessage()    {}
+func (*SlashRecord) Descriptor() ([]byte, []int) {
+	return fileDescriptor_5b27a82b276e7f93, []int{4}
+}
+func (m *SlashRecord) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *SlashRecord) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_SlashRecord.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *SlashRecord) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_SlashRecord.Merge(m, src)
+}
+func (m *SlashRecord) XXX_Size() int {
+	return m.Size()
+}
+func (m *SlashRecord) XXX_DiscardUnknown() {
+	xxx_messageInfo_SlashRecord.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_SlashRecord proto.InternalMessageInfo
+
+func (m *SlashRecord) GetWaitingOnReply() bool {
+	if m != nil {
+		return m.WaitingOnReply
+	}
+	return false
+}
+
+func (m *SlashRecord) GetSendTime() time.Time {
+	if m != nil {
+		return m.SendTime
+	}
+	return time.Time{}
+}
+
 func init() {
 	proto.RegisterType((*Params)(nil), "interchain_security.ccv.consumer.v1.Params")
 	proto.RegisterType((*LastTransmissionBlockHeight)(nil), "interchain_security.ccv.consumer.v1.LastTransmissionBlockHeight")
 	proto.RegisterType((*CrossChainValidator)(nil), "interchain_security.ccv.consumer.v1.CrossChainValidator")
 	proto.RegisterType((*MaturingVSCPacket)(nil), "interchain_security.ccv.consumer.v1.MaturingVSCPacket")
+	proto.RegisterType((*SlashRecord)(nil), "interchain_security.ccv.consumer.v1.SlashRecord")
 }
 
 func init() {
@@ -366,57 +421,60 @@ func init() {
 }
 
 var fileDescriptor_5b27a82b276e7f93 = []byte{
-	// 786 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x54, 0xcf, 0x6e, 0xdb, 0x36,
-	0x18, 0x8f, 0x96, 0xd6, 0x4d, 0x98, 0x14, 0x6b, 0x59, 0x2f, 0x55, 0x33, 0x40, 0x76, 0xdd, 0x1e,
-	0x7c, 0x89, 0x84, 0x26, 0xdb, 0xa5, 0xc0, 0x0e, 0xb5, 0xb3, 0xa2, 0xdd, 0xbf, 0x78, 0xaa, 0xd1,
-	0x01, 0xdb, 0x81, 0xa0, 0x28, 0x5a, 0x22, 0x22, 0x91, 0x02, 0x49, 0xa9, 0xd3, 0x7d, 0x0f, 0xd0,
-	0xe3, 0x1e, 0x61, 0x0f, 0xb0, 0x87, 0x28, 0x76, 0xea, 0x71, 0xa7, 0x6e, 0x48, 0xde, 0x60, 0x4f,
-	0x30, 0x90, 0x92, 0x5c, 0x3b, 0x6d, 0x80, 0xdc, 0xf8, 0xe9, 0xf7, 0xfb, 0x7e, 0xfa, 0xfe, 0x83,
-	0x43, 0xc6, 0x35, 0x95, 0x24, 0xc5, 0x8c, 0x23, 0x45, 0x49, 0x29, 0x99, 0xae, 0x03, 0x42, 0xaa,
-	0x80, 0x08, 0xae, 0xca, 0x9c, 0xca, 0xa0, 0x7a, 0xb4, 0x7c, 0xfb, 0x85, 0x14, 0x5a, 0xc0, 0x07,
-	0x1f, 0xf1, 0xf1, 0x09, 0xa9, 0xfc, 0x25, 0xaf, 0x7a, 0xb4, 0xff, 0xf0, 0x32, 0x61, 0xa3, 0x47,
-	0xaa, 0x46, 0x6a, 0xff, 0x5e, 0x22, 0x44, 0x92, 0xd1, 0xc0, 0x5a, 0x51, 0xb9, 0x08, 0x30, 0xaf,
-	0x5b, 0xa8, 0x9f, 0x88, 0x44, 0xd8, 0x67, 0x60, 0x5e, 0x9d, 0x03, 0x11, 0x2a, 0x17, 0x0a, 0x35,
-	0x40, 0x63, 0xb4, 0x90, 0x77, 0x51, 0x2b, 0x2e, 0x25, 0xd6, 0x4c, 0xf0, 0x16, 0x1f, 0x5c, 0xc4,
-	0x35, 0xcb, 0xa9, 0xd2, 0x38, 0x2f, 0x1a, 0xc2, 0xe8, 0xb7, 0x1e, 0xe8, 0xcd, 0xb0, 0xc4, 0xb9,
-	0x82, 0x2e, 0xb8, 0x41, 0x39, 0x8e, 0x32, 0x1a, 0xbb, 0xce, 0xd0, 0x19, 0x6f, 0x85, 0x9d, 0x09,
-	0x4f, 0xc0, 0xc3, 0x28, 0x13, 0xe4, 0x54, 0xa1, 0x82, 0x4a, 0x14, 0x33, 0xa5, 0x25, 0x8b, 0x4a,
-	0xf3, 0x1b, 0xa4, 0x25, 0xe6, 0x2a, 0x67, 0x4a, 0x31, 0xc1, 0xdd, 0x4f, 0x86, 0xce, 0x78, 0x33,
-	0xbc, 0xdf, 0x70, 0x67, 0x54, 0x1e, 0xaf, 0x30, 0xe7, 0x2b, 0x44, 0xf8, 0x0d, 0xb8, 0x7f, 0xa9,
-	0x0a, 0x22, 0x29, 0xe6, 0x9c, 0x66, 0xee, 0xe6, 0xd0, 0x19, 0x6f, 0x87, 0x83, 0xf8, 0x12, 0x91,
-	0x69, 0x43, 0x83, 0x8f, 0xc1, 0x7e, 0x21, 0x45, 0xc5, 0x62, 0x2a, 0xd1, 0x82, 0x52, 0x54, 0x08,
-	0x91, 0x21, 0x1c, 0xc7, 0x12, 0x29, 0x2d, 0xdd, 0x6b, 0x56, 0x64, 0xaf, 0x63, 0x3c, 0xa5, 0x74,
-	0x26, 0x44, 0xf6, 0x24, 0x8e, 0xe5, 0x0b, 0x2d, 0xe1, 0x8f, 0x00, 0x12, 0x52, 0x21, 0x53, 0x14,
-	0x51, 0x6a, 0x93, 0x1d, 0x13, 0xb1, 0x7b, 0x7d, 0xe8, 0x8c, 0x77, 0x0e, 0xef, 0xf9, 0x4d, 0xed,
-	0xfc, 0xae, 0x76, 0xfe, 0x71, 0x5b, 0xdb, 0xc9, 0xd6, 0x9b, 0x77, 0x83, 0x8d, 0xdf, 0xff, 0x19,
-	0x38, 0xe1, 0x2d, 0x42, 0xaa, 0x79, 0xe3, 0x3d, 0xb3, 0xce, 0xf0, 0x17, 0x70, 0xd7, 0x66, 0xb3,
-	0xa0, 0xf2, 0xa2, 0x6e, 0xef, 0xea, 0xba, 0x9f, 0x75, 0x1a, 0xeb, 0xe2, 0xcf, 0xc0, 0xb0, 0x9b,
-	0x37, 0x24, 0xe9, 0x5a, 0x09, 0x17, 0x12, 0x13, 0xf3, 0x70, 0x6f, 0xd8, 0x8c, 0xbd, 0x8e, 0x17,
-	0xae, 0xd1, 0x9e, 0xb6, 0x2c, 0x78, 0x00, 0x60, 0xca, 0x94, 0x16, 0x92, 0x11, 0x9c, 0x21, 0xca,
-	0xb5, 0x64, 0x54, 0xb9, 0x5b, 0xb6, 0x81, 0xb7, 0xdf, 0x23, 0x5f, 0x37, 0x00, 0xfc, 0x01, 0xdc,
-	0x2a, 0x79, 0x24, 0x78, 0xcc, 0x78, 0xd2, 0xa5, 0xb3, 0x7d, 0xf5, 0x74, 0x3e, 0x5d, 0x3a, 0xb7,
-	0x89, 0x1c, 0x81, 0x3d, 0x25, 0x16, 0x1a, 0x89, 0x42, 0x23, 0x53, 0x21, 0x9d, 0x4a, 0xaa, 0x52,
-	0x91, 0xc5, 0x2e, 0xb0, 0xe1, 0xdf, 0x31, 0xe8, 0x49, 0xa1, 0x4f, 0x4a, 0x3d, 0xef, 0x20, 0xf8,
-	0x00, 0xdc, 0x94, 0xf4, 0x15, 0x96, 0x31, 0x8a, 0x29, 0x17, 0xb9, 0x72, 0x77, 0x86, 0x9b, 0xe3,
-	0xed, 0x70, 0xb7, 0xf9, 0x78, 0x6c, 0xbf, 0xc1, 0x2f, 0xc0, 0xb2, 0xd9, 0x68, 0x9d, 0xbd, 0x6b,
-	0xd9, 0xfd, 0x0e, 0x0d, 0x57, 0xbc, 0x46, 0x5f, 0x82, 0xcf, 0xbf, 0xc3, 0x4a, 0xaf, 0xce, 0xd7,
-	0xc4, 0x4c, 0xf1, 0x33, 0xca, 0x92, 0x54, 0xc3, 0x3d, 0xd0, 0x4b, 0xed, 0xcb, 0x6e, 0xc6, 0x66,
-	0xd8, 0x5a, 0xa3, 0x3f, 0x1c, 0x70, 0x67, 0x2a, 0x85, 0x52, 0x53, 0xb3, 0xf3, 0x2f, 0x71, 0xc6,
-	0x62, 0xac, 0x85, 0x34, 0xab, 0x64, 0x26, 0x90, 0x2a, 0x65, 0x1d, 0x76, 0xc3, 0xce, 0x84, 0x7d,
-	0x70, 0xbd, 0x10, 0xaf, 0xa8, 0x6c, 0x77, 0xa5, 0x31, 0x20, 0x06, 0xbd, 0xa2, 0x8c, 0x4e, 0x69,
-	0x6d, 0x87, 0x7e, 0xe7, 0xb0, 0xff, 0x41, 0x51, 0x9f, 0xf0, 0x7a, 0x72, 0xf4, 0xdf, 0xbb, 0xc1,
-	0xdd, 0x1a, 0xe7, 0xd9, 0xe3, 0x91, 0xe9, 0x2e, 0xe5, 0xaa, 0x54, 0xa8, 0xf1, 0x1b, 0xfd, 0xf5,
-	0xe7, 0x41, 0xbf, 0xbd, 0x0c, 0x44, 0xd6, 0x85, 0x16, 0xfe, 0xac, 0x8c, 0xbe, 0xa5, 0x75, 0xd8,
-	0x0a, 0x8f, 0x34, 0xb8, 0xfd, 0x3d, 0xd6, 0xa5, 0x64, 0x3c, 0x79, 0xf9, 0x62, 0x3a, 0xc3, 0xe4,
-	0x94, 0x6a, 0x13, 0x4d, 0xa5, 0xc8, 0xf3, 0x66, 0xe1, 0xaf, 0x85, 0x8d, 0x01, 0x9f, 0x83, 0x9b,
-	0xb9, 0xa5, 0xea, 0xda, 0x8e, 0xb0, 0x8d, 0x75, 0xe7, 0x70, 0xff, 0x83, 0xa0, 0xe6, 0xdd, 0x31,
-	0x69, 0x5a, 0xfd, 0xda, 0xb4, 0x7a, 0xb7, 0x73, 0x35, 0xe0, 0xe4, 0xa7, 0x37, 0x67, 0x9e, 0xf3,
-	0xf6, 0xcc, 0x73, 0xfe, 0x3d, 0xf3, 0x9c, 0xd7, 0xe7, 0xde, 0xc6, 0xdb, 0x73, 0x6f, 0xe3, 0xef,
-	0x73, 0x6f, 0xe3, 0xe7, 0xaf, 0x12, 0xa6, 0xd3, 0x32, 0xf2, 0x89, 0xc8, 0xdb, 0x93, 0x16, 0xbc,
-	0xbf, 0x9e, 0x07, 0xcb, 0xeb, 0x59, 0x1d, 0x05, 0xbf, 0xae, 0xdf, 0x66, 0x5d, 0x17, 0x54, 0x45,
-	0x3d, 0x1b, 0xc4, 0xd1, 0xff, 0x01, 0x00, 0x00, 0xff, 0xff, 0x26, 0xe0, 0xb8, 0xdf, 0xcc, 0x05,
-	0x00, 0x00,
+	// 836 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x55, 0xcf, 0x6e, 0xdb, 0x36,
+	0x18, 0x8f, 0x96, 0xd6, 0x4d, 0xe8, 0x74, 0x4b, 0x59, 0x2f, 0x55, 0x33, 0xc0, 0x76, 0xdd, 0x1e,
+	0x7c, 0x89, 0x8d, 0x26, 0xdb, 0xa5, 0xc0, 0x0e, 0xf9, 0xb3, 0xa2, 0xdd, 0xbf, 0x78, 0x4a, 0xd0,
+	0x01, 0xdb, 0x81, 0xa0, 0xa8, 0xcf, 0x16, 0x11, 0x89, 0x14, 0x48, 0x4a, 0x99, 0x76, 0xde, 0x03,
+	0xf4, 0xb8, 0x47, 0xd8, 0x03, 0xec, 0x21, 0x8a, 0x9d, 0x7a, 0xdc, 0xa9, 0x1b, 0x92, 0x37, 0xd8,
+	0x13, 0x0c, 0xa4, 0x24, 0xd7, 0x4e, 0x17, 0xa0, 0xbb, 0xf1, 0xe3, 0xef, 0x8f, 0xf8, 0x7d, 0xfc,
+	0xf8, 0x09, 0xed, 0x72, 0x61, 0x40, 0xb1, 0x98, 0x72, 0x41, 0x34, 0xb0, 0x5c, 0x71, 0x53, 0x8e,
+	0x19, 0x2b, 0xc6, 0x4c, 0x0a, 0x9d, 0xa7, 0xa0, 0xc6, 0xc5, 0xe3, 0xf9, 0x7a, 0x94, 0x29, 0x69,
+	0x24, 0x7e, 0xf8, 0x1f, 0x9a, 0x11, 0x63, 0xc5, 0x68, 0xce, 0x2b, 0x1e, 0x6f, 0x3f, 0xba, 0xce,
+	0xd8, 0xfa, 0xb1, 0xa2, 0xb2, 0xda, 0xbe, 0x3f, 0x93, 0x72, 0x96, 0xc0, 0xd8, 0x45, 0x61, 0x3e,
+	0x1d, 0x53, 0x51, 0xd6, 0x50, 0x67, 0x26, 0x67, 0xd2, 0x2d, 0xc7, 0x76, 0xd5, 0x08, 0x98, 0xd4,
+	0xa9, 0xd4, 0xa4, 0x02, 0xaa, 0xa0, 0x86, 0xba, 0x57, 0xbd, 0xa2, 0x5c, 0x51, 0xc3, 0xa5, 0xa8,
+	0xf1, 0xde, 0x55, 0xdc, 0xf0, 0x14, 0xb4, 0xa1, 0x69, 0x56, 0x11, 0x06, 0xbf, 0xb4, 0x50, 0x6b,
+	0x42, 0x15, 0x4d, 0x35, 0xf6, 0xd1, 0x2d, 0x10, 0x34, 0x4c, 0x20, 0xf2, 0xbd, 0xbe, 0x37, 0x5c,
+	0x0b, 0x9a, 0x10, 0x1f, 0xa3, 0x47, 0x61, 0x22, 0xd9, 0x99, 0x26, 0x19, 0x28, 0x12, 0x71, 0x6d,
+	0x14, 0x0f, 0x73, 0xfb, 0x19, 0x62, 0x14, 0x15, 0x3a, 0xe5, 0x5a, 0x73, 0x29, 0xfc, 0x0f, 0xfa,
+	0xde, 0x70, 0x35, 0x78, 0x50, 0x71, 0x27, 0xa0, 0x8e, 0x16, 0x98, 0xa7, 0x0b, 0x44, 0xfc, 0x25,
+	0x7a, 0x70, 0xad, 0x0b, 0x61, 0x31, 0x15, 0x02, 0x12, 0x7f, 0xb5, 0xef, 0x0d, 0xd7, 0x83, 0x5e,
+	0x74, 0x8d, 0xc9, 0x61, 0x45, 0xc3, 0x4f, 0xd0, 0x76, 0xa6, 0x64, 0xc1, 0x23, 0x50, 0x64, 0x0a,
+	0x40, 0x32, 0x29, 0x13, 0x42, 0xa3, 0x48, 0x11, 0x6d, 0x94, 0x7f, 0xc3, 0x99, 0x6c, 0x35, 0x8c,
+	0xa7, 0x00, 0x13, 0x29, 0x93, 0xfd, 0x28, 0x52, 0x27, 0x46, 0xe1, 0xef, 0x10, 0x66, 0xac, 0x20,
+	0xb6, 0x28, 0x32, 0x37, 0x36, 0x3b, 0x2e, 0x23, 0xff, 0x66, 0xdf, 0x1b, 0xb6, 0x77, 0xef, 0x8f,
+	0xaa, 0xda, 0x8d, 0x9a, 0xda, 0x8d, 0x8e, 0xea, 0xda, 0x1e, 0xac, 0xbd, 0x7a, 0xd3, 0x5b, 0xf9,
+	0xf5, 0xaf, 0x9e, 0x17, 0x6c, 0x32, 0x56, 0x9c, 0x56, 0xea, 0x89, 0x13, 0xe3, 0x1f, 0xd1, 0x3d,
+	0x97, 0xcd, 0x14, 0xd4, 0x55, 0xdf, 0xd6, 0xfb, 0xfb, 0x7e, 0xdc, 0x78, 0x2c, 0x9b, 0x3f, 0x43,
+	0xfd, 0xa6, 0xdf, 0x88, 0x82, 0xa5, 0x12, 0x4e, 0x15, 0x65, 0x76, 0xe1, 0xdf, 0x72, 0x19, 0x77,
+	0x1b, 0x5e, 0xb0, 0x44, 0x7b, 0x5a, 0xb3, 0xf0, 0x0e, 0xc2, 0x31, 0xd7, 0x46, 0x2a, 0xce, 0x68,
+	0x42, 0x40, 0x18, 0xc5, 0x41, 0xfb, 0x6b, 0xee, 0x02, 0xef, 0xbc, 0x45, 0xbe, 0xa8, 0x00, 0xfc,
+	0x2d, 0xda, 0xcc, 0x45, 0x28, 0x45, 0xc4, 0xc5, 0xac, 0x49, 0x67, 0xfd, 0xfd, 0xd3, 0xf9, 0x68,
+	0x2e, 0xae, 0x13, 0xd9, 0x43, 0x5b, 0x5a, 0x4e, 0x0d, 0x91, 0x99, 0x21, 0xb6, 0x42, 0x26, 0x56,
+	0xa0, 0x63, 0x99, 0x44, 0x3e, 0x72, 0xc7, 0xbf, 0x6b, 0xd1, 0xe3, 0xcc, 0x1c, 0xe7, 0xe6, 0xb4,
+	0x81, 0xf0, 0x43, 0x74, 0x5b, 0xc1, 0x39, 0x55, 0x11, 0x89, 0x40, 0xc8, 0x54, 0xfb, 0xed, 0xfe,
+	0xea, 0x70, 0x3d, 0xd8, 0xa8, 0x36, 0x8f, 0xdc, 0x1e, 0xfe, 0x14, 0xcd, 0x2f, 0x9b, 0x2c, 0xb3,
+	0x37, 0x1c, 0xbb, 0xd3, 0xa0, 0xc1, 0x82, 0x6a, 0xf0, 0x19, 0xfa, 0xe4, 0x6b, 0xaa, 0xcd, 0x62,
+	0x7f, 0x1d, 0xd8, 0x2e, 0x7e, 0x06, 0x7c, 0x16, 0x1b, 0xbc, 0x85, 0x5a, 0xb1, 0x5b, 0xb9, 0x97,
+	0xb1, 0x1a, 0xd4, 0xd1, 0xe0, 0x37, 0x0f, 0xdd, 0x3d, 0x54, 0x52, 0xeb, 0x43, 0xfb, 0xe6, 0x5f,
+	0xd0, 0x84, 0x47, 0xd4, 0x48, 0x65, 0x9f, 0x92, 0xed, 0x40, 0xd0, 0xda, 0x09, 0x36, 0x82, 0x26,
+	0xc4, 0x1d, 0x74, 0x33, 0x93, 0xe7, 0xa0, 0xea, 0xb7, 0x52, 0x05, 0x98, 0xa2, 0x56, 0x96, 0x87,
+	0x67, 0x50, 0xba, 0xa6, 0x6f, 0xef, 0x76, 0xde, 0x29, 0xea, 0xbe, 0x28, 0x0f, 0xf6, 0xfe, 0x79,
+	0xd3, 0xbb, 0x57, 0xd2, 0x34, 0x79, 0x32, 0xb0, 0xb7, 0x0b, 0x42, 0xe7, 0x9a, 0x54, 0xba, 0xc1,
+	0x1f, 0xbf, 0xef, 0x74, 0xea, 0xc9, 0xc0, 0x54, 0x99, 0x19, 0x39, 0x9a, 0xe4, 0xe1, 0x57, 0x50,
+	0x06, 0xb5, 0xf1, 0xc0, 0xa0, 0x3b, 0xdf, 0x50, 0x93, 0x2b, 0x2e, 0x66, 0x2f, 0x4e, 0x0e, 0x27,
+	0x94, 0x9d, 0x81, 0xb1, 0xa7, 0x29, 0x34, 0x7b, 0x5e, 0x3d, 0xf8, 0x1b, 0x41, 0x15, 0xe0, 0xe7,
+	0xe8, 0x76, 0xea, 0xa8, 0xa6, 0x74, 0x2d, 0xec, 0xce, 0xda, 0xde, 0xdd, 0x7e, 0xe7, 0x50, 0xa7,
+	0xcd, 0x30, 0xa9, 0xae, 0xfa, 0xa5, 0xbd, 0xea, 0x8d, 0x46, 0x6a, 0xc1, 0xc1, 0xcf, 0xa8, 0x7d,
+	0x92, 0x50, 0x1d, 0x07, 0xc0, 0xa4, 0x8a, 0xf0, 0x10, 0x6d, 0x9e, 0x53, 0x6e, 0x6c, 0x13, 0x49,
+	0x41, 0x14, 0x64, 0x49, 0x59, 0xcf, 0x9a, 0x0f, 0xeb, 0xfd, 0x63, 0x11, 0xd8, 0x5d, 0xbc, 0x8f,
+	0xd6, 0x35, 0x88, 0xe8, 0xff, 0x7f, 0x7f, 0xcd, 0xca, 0x2c, 0x70, 0xf0, 0xfd, 0xab, 0x8b, 0xae,
+	0xf7, 0xfa, 0xa2, 0xeb, 0xfd, 0x7d, 0xd1, 0xf5, 0x5e, 0x5e, 0x76, 0x57, 0x5e, 0x5f, 0x76, 0x57,
+	0xfe, 0xbc, 0xec, 0xae, 0xfc, 0xf0, 0xf9, 0x8c, 0x9b, 0x38, 0x0f, 0x47, 0x4c, 0xa6, 0xf5, 0x38,
+	0x1d, 0xbf, 0x9d, 0xdc, 0x3b, 0xf3, 0xc9, 0x5d, 0xec, 0x8d, 0x7f, 0x5a, 0xfe, 0x2f, 0x98, 0x32,
+	0x03, 0x1d, 0xb6, 0xdc, 0x01, 0xf6, 0xfe, 0x0d, 0x00, 0x00, 0xff, 0xff, 0x61, 0xb7, 0xcd, 0x97,
+	0x48, 0x06, 0x00, 0x00,
 }
 
 func (m *Params) Marshal() (dAtA []byte, err error) {
@@ -643,6 +701,47 @@ func (m *MaturingVSCPacket) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	return len(dAtA) - i, nil
 }
 
+func (m *SlashRecord) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *SlashRecord) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *SlashRecord) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	n6, err6 := github_com_cosmos_gogoproto_types.StdTimeMarshalTo(m.SendTime, dAtA[i-github_com_cosmos_gogoproto_types.SizeOfStdTime(m.SendTime):])
+	if err6 != nil {
+		return 0, err6
+	}
+	i -= n6
+	i = encodeVarintConsumer(dAtA, i, uint64(n6))
+	i--
+	dAtA[i] = 0x12
+	if m.WaitingOnReply {
+		i--
+		if m.WaitingOnReply {
+			dAtA[i] = 1
+		} else {
+			dAtA[i] = 0
+		}
+		i--
+		dAtA[i] = 0x8
+	}
+	return len(dAtA) - i, nil
+}
+
 func encodeVarintConsumer(dAtA []byte, offset int, v uint64) int {
 	offset -= sovConsumer(v)
 	base := offset
@@ -752,6 +851,20 @@ func (m *MaturingVSCPacket) Size() (n int) {
 	return n
 }
 
+func (m *SlashRecord) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	if m.WaitingOnReply {
+		n += 2
+	}
+	l = github_com_cosmos_gogoproto_types.SizeOfStdTime(m.SendTime)
+	n += 1 + l + sovConsumer(uint64(l))
+	return n
+}
+
 func sovConsumer(x uint64) (n int) {
 	return (math_bits.Len64(x|1) + 6) / 7
 }
@@ -1467,6 +1580,109 @@ func (m *MaturingVSCPacket) Unmarshal(dAtA []byte) error {
 	}
 	return nil
 }
+func (m *SlashRecord) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowConsumer
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: SlashRecord: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: SlashRecord: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field WaitingOnReply", wireType)
+			}
+			var v int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowConsumer
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				v |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			m.WaitingOnReply = bool(v != 0)
+		case 2:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field SendTime", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowConsumer
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthConsumer
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthConsumer
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if err := github_com_cosmos_gogoproto_types.StdTimeUnmarshal(&m.SendTime, dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipConsumer(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthConsumer
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
 func skipConsumer(dAtA []byte) (n int, err error) {
 	l := len(dAtA)
 	iNdEx := 0
diff --git a/x/ccv/consumer/types/keys.go b/x/ccv/consumer/types/keys.go
index 8b792419ef..b755cf9f5a 100644
--- a/x/ccv/consumer/types/keys.go
+++ b/x/ccv/consumer/types/keys.go
@@ -102,6 +102,9 @@ const (
 	// This index is used for implementing a FIFO queue of pending packets in the KV store.
 	PendingPacketsIndexByteKey
 
+	// SlashRecordByteKey is the single byte key storing the consumer's slash record.
+	SlashRecordByteKey
+
 	// NOTE: DO NOT ADD NEW BYTE PREFIXES HERE WITHOUT ADDING THEM TO getAllKeyPrefixes() IN keys_test.go
 )
 
@@ -218,6 +221,11 @@ func PendingPacketsIndexKey() []byte {
 	return []byte{PendingPacketsIndexByteKey}
 }
 
+// SlashRecordKey returns the key storing the consumer's slash record.
+func SlashRecordKey() []byte {
+	return []byte{SlashRecordByteKey}
+}
+
 // NOTE: DO	NOT ADD FULLY DEFINED KEY FUNCTIONS WITHOUT ADDING THEM TO getAllFullyDefinedKeys() IN keys_test.go
 
 //
diff --git a/x/ccv/consumer/types/keys_test.go b/x/ccv/consumer/types/keys_test.go
index 5290dd3599..a8cebee284 100644
--- a/x/ccv/consumer/types/keys_test.go
+++ b/x/ccv/consumer/types/keys_test.go
@@ -42,6 +42,7 @@ func getAllKeyPrefixes() []byte {
 		StandaloneTransferChannelIDByteKey,
 		PrevStandaloneChainByteKey,
 		PendingPacketsIndexByteKey,
+		SlashRecordByteKey,
 	}
 }
 
@@ -79,5 +80,6 @@ func getAllFullyDefinedKeys() [][]byte {
 		StandaloneTransferChannelIDKey(),
 		PrevStandaloneChainKey(),
 		PendingPacketsIndexKey(),
+		SlashRecordKey(),
 	}
 }
diff --git a/x/ccv/consumer/types/throttle_retry.go b/x/ccv/consumer/types/throttle_retry.go
new file mode 100644
index 0000000000..9ea179ffe4
--- /dev/null
+++ b/x/ccv/consumer/types/throttle_retry.go
@@ -0,0 +1,11 @@
+package types
+
+import time "time"
+
+// NewSlashRecord creates a new slash record
+func NewSlashRecord(sendTime time.Time, waitingOnReply bool) (record SlashRecord) {
+	return SlashRecord{
+		SendTime:       sendTime,
+		WaitingOnReply: true,
+	}
+}
diff --git a/x/ccv/provider/keeper/relay.go b/x/ccv/provider/keeper/relay.go
index df4fdb98ce..d63594dad1 100644
--- a/x/ccv/provider/keeper/relay.go
+++ b/x/ccv/provider/keeper/relay.go
@@ -44,7 +44,7 @@ func (k Keeper) OnRecvVSCMaturedPacket(
 		"vscID", data.ValsetUpdateId,
 	)
 
-	ack := channeltypes.NewResultAcknowledgement([]byte{byte(1)})
+	ack := channeltypes.NewResultAcknowledgement(ccv.V1Result)
 	return ack
 }
 
@@ -355,7 +355,7 @@ func (k Keeper) OnRecvSlashPacket(ctx sdk.Context, packet channeltypes.Packet, d
 
 		// return successful ack, as an error would result
 		// in the consumer closing the CCV channel
-		return channeltypes.NewResultAcknowledgement([]byte{byte(1)})
+		return channeltypes.NewResultAcknowledgement(ccv.V1Result)
 	}
 
 	// Queue a slash entry to the global queue, which will be seen by the throttling logic
@@ -379,7 +379,7 @@ func (k Keeper) OnRecvSlashPacket(ctx sdk.Context, packet channeltypes.Packet, d
 		"infractionType", data.Infraction,
 	)
 
-	return channeltypes.NewResultAcknowledgement([]byte{byte(1)})
+	return channeltypes.NewResultAcknowledgement(ccv.V1Result)
 }
 
 // ValidateSlashPacket validates a recv slash packet before it is
diff --git a/x/ccv/types/ccv.go b/x/ccv/types/ccv.go
index 91175eab5d..428f726824 100644
--- a/x/ccv/types/ccv.go
+++ b/x/ccv/types/ccv.go
@@ -165,6 +165,20 @@ func (vdt1 SlashPacketDataV1) FromV1() *SlashPacketData {
 	}
 }
 
+type PacketAckResult []byte
+
+var ( // slice types can't be const
+
+	// The result ack that has historically been sent from the provider.
+	// A provider with v1 throttling sends these acks for all successfully recv packets.
+	V1Result = PacketAckResult([]byte{byte(1)})
+	// Slash packet handled result ack, sent by a throttling v2 provider to indicate that a slash packet was handled.
+	SlashPacketHandledResult = PacketAckResult([]byte{byte(2)})
+	// Slash packet bounced result ack, sent by a throttling v2 provider to indicate that a slash packet was NOT handled
+	// and should eventually be retried.
+	SlashPacketBouncedResult = PacketAckResult([]byte{byte(3)})
+)
+
 // An exported wrapper around the auto generated isConsumerPacketData_Data interface, only for
 // AppendPendingPacket to accept the interface as an argument.
 type ExportedIsConsumerPacketData_Data interface {