Switch to compounding when consolidating with source==target (#14511)

* Switch to compounding when consolidating with source==target

* Feedback

CHANGELOG.md

@@ -61,6 +61,7 @@ The format is based on Keep a Changelog, and this project adheres to Semantic Ve
 - Moved `ConvertKzgCommitmentToVersionedHash` to the `primitives` package.
 - reversed the boolean return on `BatchVerifyDepositsSignatures`, from need verification, to all keys successfully verified
 - Fix `engine_exchangeCapabilities` implementation.
+- Switch to compounding when consolidating with source==target.
 
 ### Deprecated
 - `--disable-grpc-gateway` flag is deprecated due to grpc gateway removal.

beacon-chain/core/electra/consolidations.go

@@ -15,6 +15,7 @@ import (
 	enginev1 "github.com/prysmaticlabs/prysm/v5/proto/engine/v1"
 	eth "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v5/time/slots"
+	log "github.com/sirupsen/logrus"
 )
 
 // ProcessPendingConsolidations implements the spec definition below. This method makes mutating
@@ -22,25 +23,28 @@ import (
 //
 // Spec definition:
 //
-//	def process_pending_consolidations(state: BeaconState) -> None:
-//	    next_pending_consolidation = 0
-//	    for pending_consolidation in state.pending_consolidations:
-//	        source_validator = state.validators[pending_consolidation.source_index]
-//	        if source_validator.slashed:
-//	            next_pending_consolidation += 1
-//	            continue
-//	        if source_validator.withdrawable_epoch > get_current_epoch(state):
-//	            break
+// def process_pending_consolidations(state: BeaconState) -> None:
 //
-//	        # Churn any target excess active balance of target and raise its max
-//	        switch_to_compounding_validator(state, pending_consolidation.target_index)
-//	        # Move active balance to target. Excess balance is withdrawable.
-//	        active_balance = get_active_balance(state, pending_consolidation.source_index)
-//	        decrease_balance(state, pending_consolidation.source_index, active_balance)
-//	        increase_balance(state, pending_consolidation.target_index, active_balance)
+//	next_epoch = Epoch(get_current_epoch(state) + 1)
+//	next_pending_consolidation = 0
+//	for pending_consolidation in state.pending_consolidations:
+//	    source_validator = state.validators[pending_consolidation.source_index]
+//	    if source_validator.slashed:
 //	        next_pending_consolidation += 1
+//	        continue
+//	    if source_validator.withdrawable_epoch > next_epoch:
+//	        break
 //
-//	    state.pending_consolidations = state.pending_consolidations[next_pending_consolidation:]
+//	    # Calculate the consolidated balance
+//	    max_effective_balance = get_max_effective_balance(source_validator)
+//	    source_effective_balance = min(state.balances[pending_consolidation.source_index], max_effective_balance)
+//
+//	    # Move active balance to target. Excess balance is withdrawable.
+//	    decrease_balance(state, pending_consolidation.source_index, source_effective_balance)
+//	    increase_balance(state, pending_consolidation.target_index, source_effective_balance)
+//	    next_pending_consolidation += 1
+//
+//	state.pending_consolidations = state.pending_consolidations[next_pending_consolidation:]
 func ProcessPendingConsolidations(ctx context.Context, st state.BeaconState) error {
 	_, span := trace.StartSpan(ctx, "electra.ProcessPendingConsolidations")
 	defer span.End()
@@ -51,12 +55,11 @@ func ProcessPendingConsolidations(ctx context.Context, st state.BeaconState) err
 
 	nextEpoch := slots.ToEpoch(st.Slot()) + 1
 
-	var nextPendingConsolidation uint64
 	pendingConsolidations, err := st.PendingConsolidations()
 	if err != nil {
 		return err
 	}
-
+	var nextPendingConsolidation uint64
 	for _, pc := range pendingConsolidations {
 		sourceValidator, err := st.ValidatorAtIndex(pc.SourceIndex)
 		if err != nil {
@@ -70,18 +73,16 @@ func ProcessPendingConsolidations(ctx context.Context, st state.BeaconState) err
 			break
 		}
 
-		if err := SwitchToCompoundingValidator(st, pc.TargetIndex); err != nil {
-			return err
-		}
-
-		activeBalance, err := st.ActiveBalanceAtIndex(pc.SourceIndex)
+		validatorBalance, err := st.BalanceAtIndex(pc.SourceIndex)
 		if err != nil {
 			return err
 		}
-		if err := helpers.DecreaseBalance(st, pc.SourceIndex, activeBalance); err != nil {
+		b := min(validatorBalance, helpers.ValidatorMaxEffectiveBalance(sourceValidator))
+
+		if err := helpers.DecreaseBalance(st, pc.SourceIndex, b); err != nil {
 			return err
 		}
-		if err := helpers.IncreaseBalance(st, pc.TargetIndex, activeBalance); err != nil {
+		if err := helpers.IncreaseBalance(st, pc.TargetIndex, b); err != nil {
 			return err
 		}
 		nextPendingConsolidation++
@@ -101,6 +102,16 @@ func ProcessPendingConsolidations(ctx context.Context, st state.BeaconState) err
 //	    state: BeaconState,
 //	    consolidation_request: ConsolidationRequest
 //	) -> None:
+//	    if is_valid_switch_to_compounding_request(state, consolidation_request):
+//	        validator_pubkeys = [v.pubkey for v in state.validators]
+//	        request_source_pubkey = consolidation_request.source_pubkey
+//	        source_index = ValidatorIndex(validator_pubkeys.index(request_source_pubkey))
+//	        switch_to_compounding_validator(state, source_index)
+//	        return
+//
+//	    # Verify that source != target, so a consolidation cannot be used as an exit.
+//	    if consolidation_request.source_pubkey == consolidation_request.target_pubkey:
+//	        return
 //	    # If the pending consolidations queue is full, consolidation requests are ignored
 //	    if len(state.pending_consolidations) == PENDING_CONSOLIDATIONS_LIMIT:
 //	        return
@@ -121,10 +132,6 @@ func ProcessPendingConsolidations(ctx context.Context, st state.BeaconState) err
 //	    source_validator = state.validators[source_index]
 //	    target_validator = state.validators[target_index]
 //
-//	    # Verify that source != target, so a consolidation cannot be used as an exit.
-//	    if source_index == target_index:
-//	        return
-//
 //	    # Verify source withdrawal credentials
 //	    has_correct_credential = has_execution_withdrawal_credential(source_validator)
 //	    is_correct_source_address = (
@@ -160,19 +167,14 @@ func ProcessPendingConsolidations(ctx context.Context, st state.BeaconState) err
 //	        source_index=source_index,
 //	        target_index=target_index
 //	    ))
+//
+//	    # Churn any target excess active balance of target and raise its max
+//	    if has_eth1_withdrawal_credential(target_validator):
+//	        switch_to_compounding_validator(state, target_index)
 func ProcessConsolidationRequests(ctx context.Context, st state.BeaconState, reqs []*enginev1.ConsolidationRequest) error {
 	if len(reqs) == 0 || st == nil {
 		return nil
 	}
-
-	activeBal, err := helpers.TotalActiveBalance(st)
-	if err != nil {
-		return err
-	}
-	churnLimit := helpers.ConsolidationChurnLimit(primitives.Gwei(activeBal))
-	if churnLimit <= primitives.Gwei(params.BeaconConfig().MinActivationBalance) {
-		return nil
-	}
 	curEpoch := slots.ToEpoch(st.Slot())
 	ffe := params.BeaconConfig().FarFutureEpoch
 	minValWithdrawDelay := params.BeaconConfig().MinValidatorWithdrawabilityDelay
@@ -182,22 +184,44 @@ func ProcessConsolidationRequests(ctx context.Context, st state.BeaconState, req
 		if ctx.Err() != nil {
 			return fmt.Errorf("cannot process consolidation requests: %w", ctx.Err())
 		}
+		if IsValidSwitchToCompoundingRequest(st, cr) {
+			srcIdx, ok := st.ValidatorIndexByPubkey(bytesutil.ToBytes48(cr.SourcePubkey))
+			if !ok {
+				log.Error("failed to find source validator index")
+				continue
+			}
+			if err := SwitchToCompoundingValidator(st, srcIdx); err != nil {
+				log.WithError(err).Error("failed to switch to compounding validator")
+			}
+			continue
+		}
+		sourcePubkey := bytesutil.ToBytes48(cr.SourcePubkey)
+		targetPubkey := bytesutil.ToBytes48(cr.TargetPubkey)
+		if sourcePubkey == targetPubkey {
+			continue
+		}
+
 		if npc, err := st.NumPendingConsolidations(); err != nil {
 			return fmt.Errorf("failed to fetch number of pending consolidations: %w", err) // This should never happen.
 		} else if npc >= pcLimit {
 			return nil
 		}
 
-		srcIdx, ok := st.ValidatorIndexByPubkey(bytesutil.ToBytes48(cr.SourcePubkey))
-		if !ok {
-			continue
+		activeBal, err := helpers.TotalActiveBalance(st)
+		if err != nil {
+			return err
 		}
-		tgtIdx, ok := st.ValidatorIndexByPubkey(bytesutil.ToBytes48(cr.TargetPubkey))
-		if !ok {
-			continue
+		churnLimit := helpers.ConsolidationChurnLimit(primitives.Gwei(activeBal))
+		if churnLimit <= primitives.Gwei(params.BeaconConfig().MinActivationBalance) {
+			return nil
 		}
 
-		if srcIdx == tgtIdx {
+		srcIdx, ok := st.ValidatorIndexByPubkey(sourcePubkey)
+		if !ok {
+			continue
+		}
+		tgtIdx, ok := st.ValidatorIndexByPubkey(targetPubkey)
+		if !ok {
 			continue
 		}
 
@@ -237,7 +261,8 @@ func ProcessConsolidationRequests(ctx context.Context, st state.BeaconState, req
 		// Initiate the exit of the source validator.
 		exitEpoch, err := ComputeConsolidationEpochAndUpdateChurn(ctx, st, primitives.Gwei(srcV.EffectiveBalance))
 		if err != nil {
-			return fmt.Errorf("failed to compute consolidaiton epoch: %w", err)
+			log.WithError(err).Error("failed to compute consolidation epoch")
+			continue
 		}
 		srcV.ExitEpoch = exitEpoch
 		srcV.WithdrawableEpoch = exitEpoch + minValWithdrawDelay
@@ -248,7 +273,95 @@ func ProcessConsolidationRequests(ctx context.Context, st state.BeaconState, req
 		if err := st.AppendPendingConsolidation(&eth.PendingConsolidation{SourceIndex: srcIdx, TargetIndex: tgtIdx}); err != nil {
 			return fmt.Errorf("failed to append pending consolidation: %w", err) // This should never happen.
 		}
+
+		if helpers.HasETH1WithdrawalCredential(tgtV) {
+			if err := SwitchToCompoundingValidator(st, tgtIdx); err != nil {
+				log.WithError(err).Error("failed to switch to compounding validator")
+				continue
+			}
+		}
 	}
 
 	return nil
 }
+
+// IsValidSwitchToCompoundingRequest returns true if the given consolidation request is valid for switching to compounding.
+//
+// Spec code:
+//
+// def is_valid_switch_to_compounding_request(
+//
+//	state: BeaconState,
+//	consolidation_request: ConsolidationRequest
+//
+// ) -> bool:
+//
+//	# Switch to compounding requires source and target be equal
+//	if consolidation_request.source_pubkey != consolidation_request.target_pubkey:
+//	    return False
+//
+//	# Verify pubkey exists
+//	source_pubkey = consolidation_request.source_pubkey
+//	validator_pubkeys = [v.pubkey for v in state.validators]
+//	if source_pubkey not in validator_pubkeys:
+//	    return False
+//
+//	source_validator = state.validators[ValidatorIndex(validator_pubkeys.index(source_pubkey))]
+//
+//	# Verify request has been authorized
+//	if source_validator.withdrawal_credentials[12:] != consolidation_request.source_address:
+//	    return False
+//
+//	# Verify source withdrawal credentials
+//	if not has_eth1_withdrawal_credential(source_validator):
+//	    return False
+//
+//	# Verify the source is active
+//	current_epoch = get_current_epoch(state)
+//	if not is_active_validator(source_validator, current_epoch):
+//	    return False
+//
+//	# Verify exit for source has not been initiated
+//	if source_validator.exit_epoch != FAR_FUTURE_EPOCH:
+//	    return False
+//
+//	return True
+func IsValidSwitchToCompoundingRequest(st state.BeaconState, req *enginev1.ConsolidationRequest) bool {
+	if req.SourcePubkey == nil || req.TargetPubkey == nil {
+		return false
+	}
+
+	if !bytes.Equal(req.SourcePubkey, req.TargetPubkey) {
+		return false
+	}
+
+	srcIdx, ok := st.ValidatorIndexByPubkey(bytesutil.ToBytes48(req.SourcePubkey))
+	if !ok {
+		return false
+	}
+	// As per the consensus specification, this error is not considered an assertion.
+	// Therefore, if the source_pubkey is not found in validator_pubkeys, we simply return false.
+	srcV, err := st.ValidatorAtIndexReadOnly(srcIdx)
+	if err != nil {
+		return false
+	}
+	sourceAddress := req.SourceAddress
+	withdrawalCreds := srcV.GetWithdrawalCredentials()
+	if len(withdrawalCreds) != 32 || len(sourceAddress) != 20 || !bytes.HasSuffix(withdrawalCreds, sourceAddress) {
+		return false
+	}
+
+	if !helpers.HasETH1WithdrawalCredential(srcV) {
+		return false
+	}
+
+	curEpoch := slots.ToEpoch(st.Slot())
+	if !helpers.IsActiveValidatorUsingTrie(srcV, curEpoch) {
+		return false
+	}
+
+	if srcV.ExitEpoch() != params.BeaconConfig().FarFutureEpoch {
+		return false
+	}
+	return true
+}

beacon-chain/core/electra/consolidations_test.go

@@ -13,6 +13,7 @@ import (
 	enginev1 "github.com/prysmaticlabs/prysm/v5/proto/engine/v1"
 	eth "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
 	"github.com/prysmaticlabs/prysm/v5/testing/require"
+	"github.com/prysmaticlabs/prysm/v5/testing/util"
 )
 
 func TestProcessPendingConsolidations(t *testing.T) {
@@ -80,10 +81,10 @@ func TestProcessPendingConsolidations(t *testing.T) {
 				require.NoError(t, err)
 				require.Equal(t, uint64(0), num)
 
-				// v1 is switched to compounding validator.
+				// v1 withdrawal credentials should not be updated.
 				v1, err := st.ValidatorAtIndex(1)
 				require.NoError(t, err)
-				require.Equal(t, params.BeaconConfig().CompoundingWithdrawalPrefixByte, v1.WithdrawalCredentials[0])
+				require.Equal(t, params.BeaconConfig().ETH1AddressWithdrawalPrefixByte, v1.WithdrawalCredentials[0])
 			},
 			wantErr: false,
 		},
@@ -396,3 +397,87 @@ func TestProcessConsolidationRequests(t *testing.T) {
 		})
 	}
 }
+
+func TestIsValidSwitchToCompoundingRequest(t *testing.T) {
+	st, _ := util.DeterministicGenesisStateElectra(t, 1)
+	t.Run("nil source pubkey", func(t *testing.T) {
+		ok := electra.IsValidSwitchToCompoundingRequest(st, &enginev1.ConsolidationRequest{
+			SourcePubkey: nil,
+			TargetPubkey: []byte{'a'},
+		})
+		require.Equal(t, false, ok)
+	})
+	t.Run("nil target pubkey", func(t *testing.T) {
+		ok := electra.IsValidSwitchToCompoundingRequest(st, &enginev1.ConsolidationRequest{
+			TargetPubkey: nil,
+			SourcePubkey: []byte{'a'},
+		})
+		require.Equal(t, false, ok)
+	})
+	t.Run("different source and target pubkey", func(t *testing.T) {
+		ok := electra.IsValidSwitchToCompoundingRequest(st, &enginev1.ConsolidationRequest{
+			TargetPubkey: []byte{'a'},
+			SourcePubkey: []byte{'b'},
+		})
+		require.Equal(t, false, ok)
+	})
+	t.Run("source validator not found in state", func(t *testing.T) {
+		ok := electra.IsValidSwitchToCompoundingRequest(st, &enginev1.ConsolidationRequest{
+			SourceAddress: make([]byte, 20),
+			TargetPubkey:  []byte{'a'},
+			SourcePubkey:  []byte{'a'},
+		})
+		require.Equal(t, false, ok)
+	})
+	t.Run("incorrect source address", func(t *testing.T) {
+		v, err := st.ValidatorAtIndex(0)
+		require.NoError(t, err)
+		pubkey := v.PublicKey
+		ok := electra.IsValidSwitchToCompoundingRequest(st, &enginev1.ConsolidationRequest{
+			SourceAddress: make([]byte, 20),
+			TargetPubkey:  pubkey,
+			SourcePubkey:  pubkey,
+		})
+		require.Equal(t, false, ok)
+	})
+	t.Run("incorrect eth1 withdrawal credential", func(t *testing.T) {
+		v, err := st.ValidatorAtIndex(0)
+		require.NoError(t, err)
+		pubkey := v.PublicKey
+		wc := v.WithdrawalCredentials
+		ok := electra.IsValidSwitchToCompoundingRequest(st, &enginev1.ConsolidationRequest{
+			SourceAddress: wc[12:],
+			TargetPubkey:  pubkey,
+			SourcePubkey:  pubkey,
+		})
+		require.Equal(t, false, ok)
+	})
+	t.Run("is valid compounding request", func(t *testing.T) {
+		v, err := st.ValidatorAtIndex(0)
+		require.NoError(t, err)
+		pubkey := v.PublicKey
+		wc := v.WithdrawalCredentials
+		v.WithdrawalCredentials[0] = 1
+		require.NoError(t, st.UpdateValidatorAtIndex(0, v))
+		ok := electra.IsValidSwitchToCompoundingRequest(st, &enginev1.ConsolidationRequest{
+			SourceAddress: wc[12:],
+			TargetPubkey:  pubkey,
+			SourcePubkey:  pubkey,
+		})
+		require.Equal(t, true, ok)
+	})
+	t.Run("already has an exit epoch", func(t *testing.T) {
+		v, err := st.ValidatorAtIndex(0)
+		require.NoError(t, err)
+		pubkey := v.PublicKey
+		wc := v.WithdrawalCredentials
+		v.ExitEpoch = 100
+		require.NoError(t, st.UpdateValidatorAtIndex(0, v))
+		ok := electra.IsValidSwitchToCompoundingRequest(st, &enginev1.ConsolidationRequest{
+			SourceAddress: wc[12:],
+			TargetPubkey:  pubkey,
+			SourcePubkey:  pubkey,
+		})
+		require.Equal(t, false, ok)
+	})
+}

beacon-chain/core/electra/validator.go

@@ -3,7 +3,6 @@ package electra
 import (
 	"errors"
 
-	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/helpers"
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/state"
 	"github.com/prysmaticlabs/prysm/v5/config/params"
 	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
@@ -18,9 +17,8 @@ import (
 // def switch_to_compounding_validator(state: BeaconState, index: ValidatorIndex) -> None:
 //
 //	validator = state.validators[index]
-//	if has_eth1_withdrawal_credential(validator):
-//		validator.withdrawal_credentials = COMPOUNDING_WITHDRAWAL_PREFIX + validator.withdrawal_credentials[1:]
-//		queue_excess_active_balance(state, index)
+//	validator.withdrawal_credentials = COMPOUNDING_WITHDRAWAL_PREFIX + validator.withdrawal_credentials[1:]
+//	queue_excess_active_balance(state, index)
 func SwitchToCompoundingValidator(s state.BeaconState, idx primitives.ValidatorIndex) error {
 	v, err := s.ValidatorAtIndex(idx)
 	if err != nil {
@@ -29,14 +27,12 @@ func SwitchToCompoundingValidator(s state.BeaconState, idx primitives.ValidatorI
 	if len(v.WithdrawalCredentials) == 0 {
 		return errors.New("validator has no withdrawal credentials")
 	}
-	if helpers.HasETH1WithdrawalCredential(v) {
-		v.WithdrawalCredentials[0] = params.BeaconConfig().CompoundingWithdrawalPrefixByte
-		if err := s.UpdateValidatorAtIndex(idx, v); err != nil {
-			return err
-		}
-		return QueueExcessActiveBalance(s, idx)
+
+	v.WithdrawalCredentials[0] = params.BeaconConfig().CompoundingWithdrawalPrefixByte
+	if err := s.UpdateValidatorAtIndex(idx, v); err != nil {
+		return err
 	}
-	return nil
+	return QueueExcessActiveBalance(s, idx)
 }
 
 // QueueExcessActiveBalance queues validators with balances above the min activation balance and adds to pending deposit.

beacon-chain/core/helpers/BUILD.bazel

@@ -63,6 +63,7 @@ go_test(
         "validators_test.go",
         "weak_subjectivity_test.go",
     ],
+    data = glob(["testdata/**"]),
     embed = [":go_default_library"],
     shard_count = 2,
     tags = ["CI_race_detection"],

testing/spectest/mainnet/electra/epoch_processing/pending_consolidations_test.go

@@ -7,6 +7,5 @@ import (
 )
 
 func TestMainnet_Electra_EpochProcessing_PendingConsolidations(t *testing.T) {
-	t.Skip("TODO: add back in after all spec test features are in.")
 	epoch_processing.RunPendingConsolidationsTests(t, "mainnet")
 }

testing/spectest/mainnet/electra/operations/consolidation_test.go

@@ -7,6 +7,5 @@ import (
 )
 
 func TestMainnet_Electra_Operations_Consolidation(t *testing.T) {
-	t.Skip("TODO: add back in after all spec test features are in.")
 	operations.RunConsolidationTest(t, "mainnet")
 }

testing/spectest/mainnet/electra/sanity/blocks_test.go

@@ -7,6 +7,5 @@ import (
 )
 
 func TestMainnet_Electra_Sanity_Blocks(t *testing.T) {
-	t.Skip("TODO: add back in after all spec test features are in.")
 	sanity.RunBlockProcessingTest(t, "mainnet", "sanity/blocks/pyspec_tests")
 }

testing/spectest/minimal/electra/epoch_processing/pending_consolidations_test.go

@@ -7,6 +7,5 @@ import (
 )
 
 func TestMinimal_Electra_EpochProcessing_PendingConsolidations(t *testing.T) {
-	t.Skip("TODO: add back in after all spec test features are in.")
 	epoch_processing.RunPendingConsolidationsTests(t, "minimal")
 }

testing/spectest/minimal/electra/operations/consolidation_test.go

@@ -7,6 +7,5 @@ import (
 )
 
 func TestMinimal_Electra_Operations_Consolidation(t *testing.T) {
-	t.Skip("TODO: add back in after all spec test features are in.")
 	operations.RunConsolidationTest(t, "minimal")
 }

testing/spectest/minimal/electra/sanity/blocks_test.go

@@ -7,6 +7,5 @@ import (
 )
 
 func TestMinimal_Electra_Sanity_Blocks(t *testing.T) {
-	t.Skip("TODO: add back in after all spec test features are in.")
 	sanity.RunBlockProcessingTest(t, "minimal", "sanity/blocks/pyspec_tests")
 }