Merge PR cosmos#2451: implement validator queue

* unbonding redelegation queue

* address some of bez and chris review

* delete old timeslices from queue

* added Rigel's test case

* added end-time to tags

* fixed bug in staking

* removed prints

* Get -> Queue

* called Endblocker in test

* implement validator queue

* Docs and PENDING

* address federicos comments

* unexposed UnbondingToUnbonded

* no copying unbonded val array to memory

PENDING.md

@@ -69,6 +69,7 @@ BREAKING CHANGES
     * [x/staking] \#2244 staking now holds a consensus-address-index instead of a consensus-pubkey-index
     * [x/staking] \#2236 more distribution hooks for distribution
     * [x/stake] \#2394 Split up UpdateValidator into distinct state transitions applied only in EndBlock
+    * [x/stake] \#2412 Added an unbonding validator queue to EndBlock to automatically update validator.Status when finished Unbonding
 
 * Tendermint
   * Update tendermint version from v0.23.0 to v0.25.0, notable changes

docs/spec/staking/end_block.md

@@ -1,5 +1,20 @@
 # End-Block 
 
+## Unbonding Validator Queue
+
+For all unbonding validators that have finished their unbonding period, this switches their validator.Status
+from sdk.Unbonding to sdk.Unbonded
+
+```golang
+validatorQueue(currTime time.Time):
+    // unbonding validators are in ordered queue from oldest to newest
+    for all unbondingValidators whose CompleteTime < currTime:
+        validator = GetValidator(unbondingValidator.ValidatorAddr)
+        validator.Status = sdk.Bonded
+        SetValidator(unbondingValidator)
+    return
+```
+
 ## Validator Set Changes
 
 The Tendermint validator set may be updated by state transitions that run at

types/context.go

@@ -188,7 +188,9 @@ func (c Context) WithBlockTime(newTime time.Time) Context {
 }
 
 func (c Context) WithBlockHeight(height int64) Context {
-	return c.withValue(contextKeyBlockHeight, height)
+	newHeader := c.BlockHeader()
+	newHeader.Height = height
+	return c.withValue(contextKeyBlockHeight, height).withValue(contextKeyBlockHeader, newHeader)
 }
 
 func (c Context) WithConsensusParams(params *abci.ConsensusParams) Context {

types/context_test.go

@@ -164,15 +164,16 @@ func TestContextWithCustom(t *testing.T) {
 	meter := types.NewGasMeter(10000)
 	minFees := types.Coins{types.NewInt64Coin("feeCoin", 1)}
 
-	ctx = types.NewContext(nil, header, ischeck, logger).
+	ctx = types.NewContext(nil, header, ischeck, logger)
+	require.Equal(t, header, ctx.BlockHeader())
+
+	ctx = ctx.
 		WithBlockHeight(height).
 		WithChainID(chainid).
 		WithTxBytes(txbytes).
 		WithVoteInfos(voteinfos).
 		WithGasMeter(meter).
 		WithMinimumFees(minFees)
-
-	require.Equal(t, header, ctx.BlockHeader())
 	require.Equal(t, height, ctx.BlockHeight())
 	require.Equal(t, chainid, ctx.ChainID())
 	require.Equal(t, ischeck, ctx.IsCheckTx())

x/stake/handler.go

@@ -35,6 +35,8 @@ func NewHandler(k keeper.Keeper) sdk.Handler {
 func EndBlocker(ctx sdk.Context, k keeper.Keeper) (ValidatorUpdates []abci.ValidatorUpdate) {
 	endBlockerTags := sdk.EmptyTags()
 
+	k.UnbondAllMatureValidatorQueue(ctx)
+
 	matureUnbonds := k.DequeueAllMatureUnbondingQueue(ctx, ctx.BlockHeader().Time)
 	for _, dvPair := range matureUnbonds {
 		err := k.CompleteUnbonding(ctx, dvPair.DelegatorAddr, dvPair.ValidatorAddr)

x/stake/handler_test.go

@@ -629,6 +629,56 @@ func TestJailValidator(t *testing.T) {
 	require.True(t, got.IsOK(), "expected ok, got %v", got)
 }
 
+func TestValidatorQueue(t *testing.T) {
+	ctx, _, keeper := keep.CreateTestInput(t, false, 1000)
+	validatorAddr, delegatorAddr := sdk.ValAddress(keep.Addrs[0]), keep.Addrs[1]
+
+	// set the unbonding time
+	params := keeper.GetParams(ctx)
+	params.UnbondingTime = 7 * time.Second
+	keeper.SetParams(ctx, params)
+
+	// create the validator
+	msgCreateValidator := newTestMsgCreateValidator(validatorAddr, keep.PKs[0], 10)
+	got := handleMsgCreateValidator(ctx, msgCreateValidator, keeper)
+	require.True(t, got.IsOK(), "expected no error on runMsgCreateValidator")
+
+	// bond a delegator
+	msgDelegate := newTestMsgDelegate(delegatorAddr, validatorAddr, 10)
+	got = handleMsgDelegate(ctx, msgDelegate, keeper)
+	require.True(t, got.IsOK(), "expected ok, got %v", got)
+
+	EndBlocker(ctx, keeper)
+
+	// unbond the all self-delegation to put validator in unbonding state
+	msgBeginUnbondingValidator := NewMsgBeginUnbonding(sdk.AccAddress(validatorAddr), validatorAddr, sdk.NewDec(10))
+	got = handleMsgBeginUnbonding(ctx, msgBeginUnbondingValidator, keeper)
+	require.True(t, got.IsOK(), "expected no error: %v", got)
+	var finishTime time.Time
+	types.MsgCdc.MustUnmarshalBinary(got.Data, &finishTime)
+	ctx = ctx.WithBlockTime(finishTime)
+	EndBlocker(ctx, keeper)
+	origHeader := ctx.BlockHeader()
+
+	validator, found := keeper.GetValidator(ctx, validatorAddr)
+	require.True(t, found)
+	require.True(t, validator.GetStatus() == sdk.Unbonding, "%v", validator)
+
+	// should still be unbonding at time 6 seconds later
+	ctx = ctx.WithBlockTime(origHeader.Time.Add(time.Second * 6))
+	EndBlocker(ctx, keeper)
+	validator, found = keeper.GetValidator(ctx, validatorAddr)
+	require.True(t, found)
+	require.True(t, validator.GetStatus() == sdk.Unbonding, "%v", validator)
+
+	// should be in unbonded state at time 7 seconds later
+	ctx = ctx.WithBlockTime(origHeader.Time.Add(time.Second * 7))
+	EndBlocker(ctx, keeper)
+	validator, found = keeper.GetValidator(ctx, validatorAddr)
+	require.True(t, found)
+	require.True(t, validator.GetStatus() == sdk.Unbonded, "%v", validator)
+}
+
 func TestUnbondingPeriod(t *testing.T) {
 	ctx, _, keeper := keep.CreateTestInput(t, false, 1000)
 	validatorAddr := sdk.ValAddress(keep.Addrs[0])

x/stake/keeper/delegation.go

@@ -447,10 +447,10 @@ func (k Keeper) getBeginInfo(ctx sdk.Context, valSrcAddr sdk.ValAddress) (
 
 		// the longest wait - just unbonding period from now
 		minTime = ctx.BlockHeader().Time.Add(k.UnbondingTime(ctx))
-		height = ctx.BlockHeader().Height
+		height = ctx.BlockHeight()
 		return minTime, height, false
 
-	case validator.IsUnbonded(ctx):
+	case validator.Status == sdk.Unbonded:
 		return minTime, height, true
 
 	case validator.Status == sdk.Unbonding:

x/stake/keeper/delegation_test.go

@@ -374,12 +374,8 @@ func TestUndelegateFromUnbondedValidator(t *testing.T) {
 	}
 	keeper.SetDelegation(ctx, delegation)
 
-	header := ctx.BlockHeader()
-	blockHeight := int64(10)
-	header.Height = blockHeight
-	blockTime := time.Unix(333, 0)
-	header.Time = blockTime
-	ctx = ctx.WithBlockHeader(header)
+	ctx = ctx.WithBlockHeight(10)
+	ctx = ctx.WithBlockTime(time.Unix(333, 0))
 
 	// unbond the all self-delegation to put validator in unbonding state
 	_, err := keeper.BeginUnbonding(ctx, val0AccAddr, addrVals[0], sdk.NewDec(10))
@@ -391,17 +387,12 @@ func TestUndelegateFromUnbondedValidator(t *testing.T) {
 
 	validator, found := keeper.GetValidator(ctx, addrVals[0])
 	require.True(t, found)
-	require.Equal(t, blockHeight, validator.UnbondingHeight)
+	require.Equal(t, ctx.BlockHeight(), validator.UnbondingHeight)
 	params := keeper.GetParams(ctx)
-	require.True(t, blockTime.Add(params.UnbondingTime).Equal(validator.UnbondingMinTime))
+	require.True(t, ctx.BlockHeader().Time.Add(params.UnbondingTime).Equal(validator.UnbondingMinTime))
 
-	// change the context to one which makes the validator considered unbonded
-	header = ctx.BlockHeader()
-	blockHeight2 := int64(20)
-	header.Height = blockHeight2
-	blockTime2 := time.Unix(444, 0).Add(params.UnbondingTime)
-	header.Time = blockTime2
-	ctx = ctx.WithBlockHeader(header)
+	// unbond the validator
+	keeper.unbondingToUnbonded(ctx, validator)
 
 	// unbond some of the other delegation's shares
 	_, err = keeper.BeginUnbonding(ctx, addrDels[0], addrVals[0], sdk.NewDec(6))
@@ -696,12 +687,8 @@ func TestRedelegateFromUnbondedValidator(t *testing.T) {
 	validator2 = testingUpdateValidator(keeper, ctx, validator2)
 	require.Equal(t, sdk.Bonded, validator2.Status)
 
-	header := ctx.BlockHeader()
-	blockHeight := int64(10)
-	header.Height = blockHeight
-	blockTime := time.Unix(333, 0)
-	header.Time = blockTime
-	ctx = ctx.WithBlockHeader(header)
+	ctx = ctx.WithBlockHeight(10)
+	ctx = ctx.WithBlockTime(time.Unix(333, 0))
 
 	// unbond the all self-delegation to put validator in unbonding state
 	_, err := keeper.BeginUnbonding(ctx, val0AccAddr, addrVals[0], sdk.NewDec(10))
@@ -713,23 +700,18 @@ func TestRedelegateFromUnbondedValidator(t *testing.T) {
 
 	validator, found := keeper.GetValidator(ctx, addrVals[0])
 	require.True(t, found)
-	require.Equal(t, blockHeight, validator.UnbondingHeight)
+	require.Equal(t, ctx.BlockHeight(), validator.UnbondingHeight)
 	params := keeper.GetParams(ctx)
-	require.True(t, blockTime.Add(params.UnbondingTime).Equal(validator.UnbondingMinTime))
+	require.True(t, ctx.BlockHeader().Time.Add(params.UnbondingTime).Equal(validator.UnbondingMinTime))
 
-	// change the context to one which makes the validator considered unbonded
-	header = ctx.BlockHeader()
-	blockHeight2 := int64(20)
-	header.Height = blockHeight2
-	blockTime2 := time.Unix(444, 0).Add(params.UnbondingTime)
-	header.Time = blockTime2
-	ctx = ctx.WithBlockHeader(header)
+	// unbond the validator
+	keeper.unbondingToUnbonded(ctx, validator)
 
-	// unbond some of the other delegation's shares
+	// redelegate some of the delegation's shares
 	_, err = keeper.BeginRedelegation(ctx, addrDels[0], addrVals[0], addrVals[1], sdk.NewDec(6))
 	require.NoError(t, err)
 
-	// no ubd should have been found, coins should have been returned direcly to account
-	ubd, found := keeper.GetRedelegation(ctx, addrDels[0], addrVals[0], addrVals[1])
-	require.False(t, found, "%v", ubd)
+	// no red should have been found
+	red, found := keeper.GetRedelegation(ctx, addrDels[0], addrVals[0], addrVals[1])
+	require.False(t, found, "%v", red)
 }

x/stake/keeper/key.go

@@ -29,6 +29,7 @@ var (
 	RedelegationByValDstIndexKey     = []byte{0x0C} // prefix for each key for an redelegation, by destination validator operator
 	UnbondingQueueKey                = []byte{0x0D} // prefix for the timestamps in unbonding queue
 	RedelegationQueueKey             = []byte{0x0E} // prefix for the timestamps in redelegations queue
+	ValidatorQueueKey                = []byte{0x0F} // prefix for the timestamps in validator queue
 )
 
 const maxDigitsForAccount = 12 // ~220,000,000 atoms created at launch
@@ -86,6 +87,12 @@ func getValidatorPowerRank(validator types.Validator) []byte {
 	return key
 }
 
+// gets the prefix for all unbonding delegations from a delegator
+func GetValidatorQueueTimeKey(timestamp time.Time) []byte {
+	bz := types.MsgCdc.MustMarshalBinary(timestamp)
+	return append(ValidatorQueueKey, bz...)
+}
+
 //______________________________________________________________________________
 
 // gets the key for delegator bond with validator

x/stake/keeper/slash.go

@@ -46,7 +46,7 @@ func (k Keeper) Slash(ctx sdk.Context, consAddr sdk.ConsAddress, infractionHeigh
 	}
 
 	// should not be slashing unbonded
-	if validator.IsUnbonded(ctx) {
+	if validator.Status == sdk.Unbonded {
 		panic(fmt.Sprintf("should not be slashing unbonded validator: %s", validator.GetOperator()))
 	}
 

x/stake/keeper/val_state_change.go

@@ -131,8 +131,9 @@ func (k Keeper) unbondedToBonded(ctx sdk.Context, validator types.Validator) typ
 	return k.bondValidator(ctx, validator)
 }
 
+// switches a validator from unbonding state to unbonded state
 func (k Keeper) unbondingToUnbonded(ctx sdk.Context, validator types.Validator) types.Validator {
-	if validator.Status != sdk.Unbonded {
+	if validator.Status != sdk.Unbonding {
 		panic(fmt.Sprintf("bad state transition unbondingToBonded, validator: %v\n", validator))
 	}
 	return k.completeUnbondingValidator(ctx, validator)
@@ -213,6 +214,9 @@ func (k Keeper) beginUnbondingValidator(ctx sdk.Context, validator types.Validat
 
 	k.SetValidatorByPowerIndex(ctx, validator, pool)
 
+	// Adds to unbonding validator queue
+	k.InsertValidatorQueue(ctx, validator)
+
 	// call the unbond hook if present
 	if k.hooks != nil {
 		k.hooks.OnValidatorBeginUnbonding(ctx, validator.ConsAddress())

x/stake/keeper/validator.go

@@ -3,6 +3,7 @@ package keeper
 import (
 	"container/list"
 	"fmt"
+	"time"
 
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	"github.com/cosmos/cosmos-sdk/x/stake/types"
@@ -281,3 +282,69 @@ func (k Keeper) GetBondedValidatorsByPower(ctx sdk.Context) []types.Validator {
 	}
 	return validators[:i] // trim
 }
+
+// gets a specific validator queue timeslice. A timeslice is a slice of ValAddresses corresponding to unbonding validators
+// that expire at a certain time.
+func (k Keeper) GetValidatorQueueTimeSlice(ctx sdk.Context, timestamp time.Time) (valAddrs []sdk.ValAddress) {
+	store := ctx.KVStore(k.storeKey)
+	bz := store.Get(GetValidatorQueueTimeKey(timestamp))
+	if bz == nil {
+		return []sdk.ValAddress{}
+	}
+	k.cdc.MustUnmarshalBinary(bz, &valAddrs)
+	return valAddrs
+}
+
+// Sets a specific validator queue timeslice.
+func (k Keeper) SetValidatorQueueTimeSlice(ctx sdk.Context, timestamp time.Time, keys []sdk.ValAddress) {
+	store := ctx.KVStore(k.storeKey)
+	bz := k.cdc.MustMarshalBinary(keys)
+	store.Set(GetValidatorQueueTimeKey(timestamp), bz)
+}
+
+// Insert an validator address to the appropriate timeslice in the validator queue
+func (k Keeper) InsertValidatorQueue(ctx sdk.Context, val types.Validator) {
+	timeSlice := k.GetValidatorQueueTimeSlice(ctx, val.UnbondingMinTime)
+	if len(timeSlice) == 0 {
+		k.SetValidatorQueueTimeSlice(ctx, val.UnbondingMinTime, []sdk.ValAddress{val.OperatorAddr})
+	} else {
+		timeSlice = append(timeSlice, val.OperatorAddr)
+		k.SetValidatorQueueTimeSlice(ctx, val.UnbondingMinTime, timeSlice)
+	}
+}
+
+// Returns all the validator queue timeslices from time 0 until endTime
+func (k Keeper) ValidatorQueueIterator(ctx sdk.Context, endTime time.Time) sdk.Iterator {
+	store := ctx.KVStore(k.storeKey)
+	return store.Iterator(ValidatorQueueKey, sdk.InclusiveEndBytes(GetValidatorQueueTimeKey(endTime)))
+}
+
+// Returns a concatenated list of all the timeslices before currTime, and deletes the timeslices from the queue
+func (k Keeper) GetAllMatureValidatorQueue(ctx sdk.Context, currTime time.Time) (matureValsAddrs []sdk.ValAddress) {
+	// gets an iterator for all timeslices from time 0 until the current Blockheader time
+	validatorTimesliceIterator := k.ValidatorQueueIterator(ctx, ctx.BlockHeader().Time)
+	for ; validatorTimesliceIterator.Valid(); validatorTimesliceIterator.Next() {
+		timeslice := []sdk.ValAddress{}
+		k.cdc.MustUnmarshalBinary(validatorTimesliceIterator.Value(), &timeslice)
+		matureValsAddrs = append(matureValsAddrs, timeslice...)
+	}
+	return matureValsAddrs
+}
+
+// Unbonds all the unbonding validators that have finished their unbonding period
+func (k Keeper) UnbondAllMatureValidatorQueue(ctx sdk.Context) {
+	store := ctx.KVStore(k.storeKey)
+	validatorTimesliceIterator := k.ValidatorQueueIterator(ctx, ctx.BlockHeader().Time)
+	for ; validatorTimesliceIterator.Valid(); validatorTimesliceIterator.Next() {
+		timeslice := []sdk.ValAddress{}
+		k.cdc.MustUnmarshalBinary(validatorTimesliceIterator.Value(), &timeslice)
+		for _, valAddr := range timeslice {
+			val, found := k.GetValidator(ctx, valAddr)
+			if !found || val.GetStatus() != sdk.Unbonding {
+				continue
+			}
+			k.unbondingToUnbonded(ctx, val)
+		}
+		store.Delete(validatorTimesliceIterator.Key())
+	}
+}

x/stake/types/validator.go

@@ -435,21 +435,6 @@ func (v Validator) BondedTokens() sdk.Dec {
 	return sdk.ZeroDec()
 }
 
-// TODO remove this once the validator queue logic is implemented
-// Returns if the validator should be considered unbonded
-func (v Validator) IsUnbonded(ctx sdk.Context) bool {
-	switch v.Status {
-	case sdk.Unbonded:
-		return true
-	case sdk.Unbonding:
-		ctxTime := ctx.BlockHeader().Time
-		if ctxTime.After(v.UnbondingMinTime) {
-			return true
-		}
-	}
-	return false
-}
-
 //______________________________________________________________________
 
 // ensure fulfills the sdk validator types