Electra: new process_consolidation_request (#14163)

* Electra: process_consolidation_request

* Enable consolidation_request spectests

beacon-chain/core/electra/churn_test.go

@@ -2,6 +2,7 @@ package electra_test
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/electra"
@@ -18,10 +19,17 @@ func createValidatorsWithTotalActiveBalance(totalBal primitives.Gwei) []*eth.Val
 	num := totalBal / primitives.Gwei(params.BeaconConfig().MinActivationBalance)
 	vals := make([]*eth.Validator, num)
 	for i := range vals {
+		wd := make([]byte, 32)
+		wd[0] = params.BeaconConfig().ETH1AddressWithdrawalPrefixByte
+		wd[31] = byte(i)
+
 		vals[i] = &eth.Validator{
-			ActivationEpoch:  primitives.Epoch(0),
-			ExitEpoch:        params.BeaconConfig().FarFutureEpoch,
-			EffectiveBalance: params.BeaconConfig().MinActivationBalance,
+			ActivationEpoch:       primitives.Epoch(0),
+			EffectiveBalance:      params.BeaconConfig().MinActivationBalance,
+			ExitEpoch:             params.BeaconConfig().FarFutureEpoch,
+			PublicKey:             []byte(fmt.Sprintf("val_%d", i)),
+			WithdrawableEpoch:     params.BeaconConfig().FarFutureEpoch,
+			WithdrawalCredentials: wd,
 		}
 	}
 	if totalBal%primitives.Gwei(params.BeaconConfig().MinActivationBalance) != 0 {

beacon-chain/core/electra/consolidations.go

@@ -1,11 +1,18 @@
 package electra
 
 import (
+	"bytes"
 	"context"
+	"fmt"
 
 	"github.com/pkg/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"
+	"github.com/prysmaticlabs/prysm/v5/encoding/bytesutil"
+	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"
 	"go.opencensus.io/trace"
 )
@@ -86,3 +93,162 @@ func ProcessPendingConsolidations(ctx context.Context, st state.BeaconState) err
 
 	return nil
 }
+
+// ProcessConsolidationRequests implements the spec definition below. This method makes mutating
+// calls to the beacon state.
+//
+//	def process_consolidation_request(
+//	    state: BeaconState,
+//	    consolidation_request: ConsolidationRequest
+//	) -> None:
+//	    # If the pending consolidations queue is full, consolidation requests are ignored
+//	    if len(state.pending_consolidations) == PENDING_CONSOLIDATIONS_LIMIT:
+//	        return
+//	    # If there is too little available consolidation churn limit, consolidation requests are ignored
+//	    if get_consolidation_churn_limit(state) <= MIN_ACTIVATION_BALANCE:
+//	        return
+//
+//	    validator_pubkeys = [v.pubkey for v in state.validators]
+//	    # Verify pubkeys exists
+//	    request_source_pubkey = consolidation_request.source_pubkey
+//	    request_target_pubkey = consolidation_request.target_pubkey
+//	    if request_source_pubkey not in validator_pubkeys:
+//	        return
+//	    if request_target_pubkey not in validator_pubkeys:
+//	        return
+//	    source_index = ValidatorIndex(validator_pubkeys.index(request_source_pubkey))
+//	    target_index = ValidatorIndex(validator_pubkeys.index(request_target_pubkey))
+//	    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 = (
+//	        source_validator.withdrawal_credentials[12:] == consolidation_request.source_address
+//	    )
+//	    if not (has_correct_credential and is_correct_source_address):
+//	        return
+//
+//	    # Verify that target has execution withdrawal credentials
+//	    if not has_execution_withdrawal_credential(target_validator):
+//	        return
+//
+//	    # Verify the source and the target are active
+//	    current_epoch = get_current_epoch(state)
+//	    if not is_active_validator(source_validator, current_epoch):
+//	        return
+//	    if not is_active_validator(target_validator, current_epoch):
+//	        return
+//	    # Verify exits for source and target have not been initiated
+//	    if source_validator.exit_epoch != FAR_FUTURE_EPOCH:
+//	        return
+//	    if target_validator.exit_epoch != FAR_FUTURE_EPOCH:
+//	        return
+//
+//	    # Initiate source validator exit and append pending consolidation
+//	    source_validator.exit_epoch = compute_consolidation_epoch_and_update_churn(
+//	        state, source_validator.effective_balance
+//	    )
+//	    source_validator.withdrawable_epoch = Epoch(
+//	        source_validator.exit_epoch + MIN_VALIDATOR_WITHDRAWABILITY_DELAY
+//	    )
+//	    state.pending_consolidations.append(PendingConsolidation(
+//	        source_index=source_index,
+//	        target_index=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
+	pcLimit := params.BeaconConfig().PendingConsolidationsLimit
+
+	for _, cr := range reqs {
+		if ctx.Err() != nil {
+			return fmt.Errorf("cannot process consolidation requests: %w", ctx.Err())
+		}
+		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
+		}
+		tgtIdx, ok := st.ValidatorIndexByPubkey(bytesutil.ToBytes48(cr.TargetPubkey))
+		if !ok {
+			continue
+		}
+
+		if srcIdx == tgtIdx {
+			continue
+		}
+
+		srcV, err := st.ValidatorAtIndex(srcIdx)
+		if err != nil {
+			return fmt.Errorf("failed to fetch source validator: %w", err) // This should never happen.
+		}
+
+		tgtV, err := st.ValidatorAtIndexReadOnly(tgtIdx)
+		if err != nil {
+			return fmt.Errorf("failed to fetch target validator: %w", err) // This should never happen.
+		}
+
+		// Verify source withdrawal credentials
+		if !helpers.HasExecutionWithdrawalCredentials(srcV) {
+			continue
+		}
+		// Confirm source_validator.withdrawal_credentials[12:] == consolidation_request.source_address
+		if len(srcV.WithdrawalCredentials) != 32 || len(cr.SourceAddress) != 20 || !bytes.HasSuffix(srcV.WithdrawalCredentials, cr.SourceAddress) {
+			continue
+		}
+
+		// Target validator must have their withdrawal credentials set appropriately.
+		if !helpers.HasExecutionWithdrawalCredentials(tgtV) {
+			continue
+		}
+
+		// Both validators must be active.
+		if !helpers.IsActiveValidator(srcV, curEpoch) || !helpers.IsActiveValidatorUsingTrie(tgtV, curEpoch) {
+			continue
+		}
+		// Neither validator are exiting.
+		if srcV.ExitEpoch != ffe || tgtV.ExitEpoch() != ffe {
+			continue
+		}
+
+		// 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)
+		}
+		srcV.ExitEpoch = exitEpoch
+		srcV.WithdrawableEpoch = exitEpoch + minValWithdrawDelay
+		if err := st.UpdateValidatorAtIndex(srcIdx, srcV); err != nil {
+			return fmt.Errorf("failed to update validator: %w", err) // This should never happen.
+		}
+
+		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.
+		}
+	}
+
+	return nil
+}

beacon-chain/core/electra/consolidations_test.go

@@ -8,6 +8,9 @@ import (
 	"github.com/prysmaticlabs/prysm/v5/beacon-chain/state"
 	state_native "github.com/prysmaticlabs/prysm/v5/beacon-chain/state/state-native"
 	"github.com/prysmaticlabs/prysm/v5/config/params"
+	"github.com/prysmaticlabs/prysm/v5/consensus-types/primitives"
+	"github.com/prysmaticlabs/prysm/v5/encoding/bytesutil"
+	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"
 )
@@ -229,3 +232,199 @@ func stateWithActiveBalanceETH(t *testing.T, balETH uint64) state.BeaconState {
 
 	return st
 }
+
+func TestProcessConsolidationRequests(t *testing.T) {
+	tests := []struct {
+		name     string
+		state    state.BeaconState
+		reqs     []*enginev1.ConsolidationRequest
+		validate func(*testing.T, state.BeaconState)
+	}{
+		{
+			name: "one valid request",
+			state: func() state.BeaconState {
+				st := &eth.BeaconStateElectra{
+					Validators: createValidatorsWithTotalActiveBalance(32000000000000000), // 32M ETH
+				}
+				// Validator scenario setup. See comments in reqs section.
+				st.Validators[3].WithdrawalCredentials = bytesutil.Bytes32(0)
+				st.Validators[8].WithdrawalCredentials = bytesutil.Bytes32(0)
+				st.Validators[9].ActivationEpoch = params.BeaconConfig().FarFutureEpoch
+				st.Validators[12].ActivationEpoch = params.BeaconConfig().FarFutureEpoch
+				st.Validators[13].ExitEpoch = 10
+				st.Validators[16].ExitEpoch = 10
+				s, err := state_native.InitializeFromProtoElectra(st)
+				require.NoError(t, err)
+				return s
+			}(),
+			reqs: []*enginev1.ConsolidationRequest{
+				// Source doesn't have withdrawal credentials.
+				{
+					SourceAddress: append(bytesutil.PadTo(nil, 19), byte(1)),
+					SourcePubkey:  []byte("val_3"),
+					TargetPubkey:  []byte("val_4"),
+				},
+				// Source withdrawal credentials don't match the consolidation address.
+				{
+					SourceAddress: append(bytesutil.PadTo(nil, 19), byte(0)), // Should be 5
+					SourcePubkey:  []byte("val_5"),
+					TargetPubkey:  []byte("val_6"),
+				},
+				// Target does not have their withdrawal credentials set appropriately.
+				{
+					SourceAddress: append(bytesutil.PadTo(nil, 19), byte(7)),
+					SourcePubkey:  []byte("val_7"),
+					TargetPubkey:  []byte("val_8"),
+				},
+				// Source is inactive.
+				{
+					SourceAddress: append(bytesutil.PadTo(nil, 19), byte(9)),
+					SourcePubkey:  []byte("val_9"),
+					TargetPubkey:  []byte("val_10"),
+				},
+				// Target is inactive.
+				{
+					SourceAddress: append(bytesutil.PadTo(nil, 19), byte(11)),
+					SourcePubkey:  []byte("val_11"),
+					TargetPubkey:  []byte("val_12"),
+				},
+				// Source is exiting.
+				{
+					SourceAddress: append(bytesutil.PadTo(nil, 19), byte(13)),
+					SourcePubkey:  []byte("val_13"),
+					TargetPubkey:  []byte("val_14"),
+				},
+				// Target is exiting.
+				{
+					SourceAddress: append(bytesutil.PadTo(nil, 19), byte(15)),
+					SourcePubkey:  []byte("val_15"),
+					TargetPubkey:  []byte("val_16"),
+				},
+				// Source doesn't exist
+				{
+					SourceAddress: append(bytesutil.PadTo(nil, 19), byte(0)),
+					SourcePubkey:  []byte("INVALID"),
+					TargetPubkey:  []byte("val_0"),
+				},
+				// Target doesn't exist
+				{
+					SourceAddress: append(bytesutil.PadTo(nil, 19), byte(0)),
+					SourcePubkey:  []byte("val_0"),
+					TargetPubkey:  []byte("INVALID"),
+				},
+				// Source == target
+				{
+					SourceAddress: append(bytesutil.PadTo(nil, 19), byte(0)),
+					SourcePubkey:  []byte("val_0"),
+					TargetPubkey:  []byte("val_0"),
+				},
+				// Valid consolidation request. This should be last to ensure invalid requests do
+				// not end the processing early.
+				{
+					SourceAddress: append(bytesutil.PadTo(nil, 19), byte(1)),
+					SourcePubkey:  []byte("val_1"),
+					TargetPubkey:  []byte("val_2"),
+				},
+			},
+			validate: func(t *testing.T, st state.BeaconState) {
+				// Verify a pending consolidation is created.
+				numPC, err := st.NumPendingConsolidations()
+				require.NoError(t, err)
+				require.Equal(t, uint64(1), numPC)
+				pcs, err := st.PendingConsolidations()
+				require.NoError(t, err)
+				require.Equal(t, primitives.ValidatorIndex(1), pcs[0].SourceIndex)
+				require.Equal(t, primitives.ValidatorIndex(2), pcs[0].TargetIndex)
+
+				// Verify the source validator is exiting.
+				src, err := st.ValidatorAtIndex(1)
+				require.NoError(t, err)
+				require.NotEqual(t, params.BeaconConfig().FarFutureEpoch, src.ExitEpoch, "source validator exit epoch not updated")
+				require.Equal(t, params.BeaconConfig().MinValidatorWithdrawabilityDelay, src.WithdrawableEpoch-src.ExitEpoch, "source validator withdrawable epoch not set correctly")
+			},
+		},
+		{
+			name: "pending consolidations limit reached",
+			state: func() state.BeaconState {
+				st := &eth.BeaconStateElectra{
+					Validators:            createValidatorsWithTotalActiveBalance(32000000000000000), // 32M ETH
+					PendingConsolidations: make([]*eth.PendingConsolidation, params.BeaconConfig().PendingConsolidationsLimit),
+				}
+				s, err := state_native.InitializeFromProtoElectra(st)
+				require.NoError(t, err)
+				return s
+			}(),
+			reqs: []*enginev1.ConsolidationRequest{
+				{
+					SourceAddress: append(bytesutil.PadTo(nil, 19), byte(1)),
+					SourcePubkey:  []byte("val_1"),
+					TargetPubkey:  []byte("val_2"),
+				},
+			},
+			validate: func(t *testing.T, st state.BeaconState) {
+				// Verify no pending consolidation is created.
+				numPC, err := st.NumPendingConsolidations()
+				require.NoError(t, err)
+				require.Equal(t, params.BeaconConfig().PendingConsolidationsLimit, numPC)
+
+				// Verify the source validator is not exiting.
+				src, err := st.ValidatorAtIndex(1)
+				require.NoError(t, err)
+				require.Equal(t, params.BeaconConfig().FarFutureEpoch, src.ExitEpoch, "source validator exit epoch should not be updated")
+				require.Equal(t, params.BeaconConfig().FarFutureEpoch, src.WithdrawableEpoch, "source validator withdrawable epoch should not be updated")
+			},
+		},
+		{
+			name: "pending consolidations limit reached during processing",
+			state: func() state.BeaconState {
+				st := &eth.BeaconStateElectra{
+					Validators:            createValidatorsWithTotalActiveBalance(32000000000000000), // 32M ETH
+					PendingConsolidations: make([]*eth.PendingConsolidation, params.BeaconConfig().PendingConsolidationsLimit-1),
+				}
+				s, err := state_native.InitializeFromProtoElectra(st)
+				require.NoError(t, err)
+				return s
+			}(),
+			reqs: []*enginev1.ConsolidationRequest{
+				{
+					SourceAddress: append(bytesutil.PadTo(nil, 19), byte(1)),
+					SourcePubkey:  []byte("val_1"),
+					TargetPubkey:  []byte("val_2"),
+				},
+				{
+					SourceAddress: append(bytesutil.PadTo(nil, 19), byte(3)),
+					SourcePubkey:  []byte("val_3"),
+					TargetPubkey:  []byte("val_4"),
+				},
+			},
+			validate: func(t *testing.T, st state.BeaconState) {
+				// Verify a pending consolidation is created.
+				numPC, err := st.NumPendingConsolidations()
+				require.NoError(t, err)
+				require.Equal(t, params.BeaconConfig().PendingConsolidationsLimit, numPC)
+
+				// The first consolidation was appended.
+				pcs, err := st.PendingConsolidations()
+				require.NoError(t, err)
+				require.Equal(t, primitives.ValidatorIndex(1), pcs[params.BeaconConfig().PendingConsolidationsLimit-1].SourceIndex)
+				require.Equal(t, primitives.ValidatorIndex(2), pcs[params.BeaconConfig().PendingConsolidationsLimit-1].TargetIndex)
+
+				// Verify the second source validator is not exiting.
+				src, err := st.ValidatorAtIndex(3)
+				require.NoError(t, err)
+				require.Equal(t, params.BeaconConfig().FarFutureEpoch, src.ExitEpoch, "source validator exit epoch should not be updated")
+				require.Equal(t, params.BeaconConfig().FarFutureEpoch, src.WithdrawableEpoch, "source validator withdrawable epoch should not be updated")
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := electra.ProcessConsolidationRequests(context.TODO(), tt.state, tt.reqs)
+			require.NoError(t, err)
+			if tt.validate != nil {
+				tt.validate(t, tt.state)
+			}
+		})
+	}
+}

beacon-chain/core/helpers/validators.go

@@ -554,7 +554,7 @@ func IsCompoundingWithdrawalCredential(creds []byte) bool {
 //	    Check if ``validator`` has a 0x01 or 0x02 prefixed withdrawal credential.
 //	    """
 //	    return has_compounding_withdrawal_credential(validator) or has_eth1_withdrawal_credential(validator)
-func HasExecutionWithdrawalCredentials(v *ethpb.Validator) bool {
+func HasExecutionWithdrawalCredentials(v interfaces.WithWithdrawalCredentials) bool {
 	if v == nil {
 		return false
 	}

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

@@ -7,6 +7,5 @@ import (
 )
 
 func TestMainnet_Electra_Operations_Consolidation(t *testing.T) {
-	t.Skip("These tests were temporarily deleted in v1.5.0-alpha.2. See https://github.com/ethereum/consensus-specs/pull/3736")
 	operations.RunConsolidationTest(t, "mainnet")
 }

testing/spectest/shared/electra/operations/consolidations.go

@@ -1,32 +1,49 @@
 package operations
 
 import (
+	"context"
 	"path"
 	"testing"
 
 	"github.com/golang/snappy"
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/core/electra"
+	"github.com/prysmaticlabs/prysm/v5/beacon-chain/state"
+	"github.com/prysmaticlabs/prysm/v5/consensus-types/interfaces"
 	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/spectest/utils"
 	"github.com/prysmaticlabs/prysm/v5/testing/util"
 )
 
 func RunConsolidationTest(t *testing.T, config string) {
-	t.Skip("Failing until spectests are updated to v1.5.0-alpha.3")
 	require.NoError(t, utils.SetConfig(t, config))
-	testFolders, testsFolderPath := utils.TestFolders(t, config, "electra", "operations/consolidation/pyspec_tests")
+	testFolders, testsFolderPath := utils.TestFolders(t, config, "electra", "operations/consolidation_request/pyspec_tests")
 	require.NotEqual(t, 0, len(testFolders), "missing tests for consolidation operation in folder")
 	for _, folder := range testFolders {
 		t.Run(folder.Name(), func(t *testing.T) {
 			folderPath := path.Join(testsFolderPath, folder.Name())
-			consolidationFile, err := util.BazelFileBytes(folderPath, "consolidation.ssz_snappy")
+			consolidationFile, err := util.BazelFileBytes(folderPath, "consolidation_request.ssz_snappy")
 			require.NoError(t, err)
 			consolidationSSZ, err := snappy.Decode(nil /* dst */, consolidationFile)
 			require.NoError(t, err, "Failed to decompress")
 			consolidation := &enginev1.ConsolidationRequest{}
 			require.NoError(t, consolidation.UnmarshalSSZ(consolidationSSZ), "Failed to unmarshal")
 
-			t.Fatal("Implement me")
+			body := &eth.BeaconBlockBodyElectra{ExecutionPayload: &enginev1.ExecutionPayloadElectra{
+				ConsolidationRequests: []*enginev1.ConsolidationRequest{consolidation},
+			}}
+			RunBlockOperationTest(t, folderPath, body, func(ctx context.Context, s state.BeaconState, b interfaces.ReadOnlySignedBeaconBlock) (state.BeaconState, error) {
+				ed, err := b.Block().Body().Execution()
+				if err != nil {
+					return nil, err
+				}
+				eed, ok := ed.(interfaces.ExecutionDataElectra)
+				if !ok {
+					t.Fatal("block does not have execution data for electra")
+				}
+				return s, electra.ProcessConsolidationRequests(ctx, s, eed.ConsolidationRequests())
+			})
 		})
 	}
 }