cl/gloas: gate fork choice on verified execution payloads (erigontech#21417)

## Summary

- Replace `HasEnvelope` (disk existence) with `verifiedExecutionPayload`
(EL returned VALID) in GLOAS fork choice logic
- Add `MarkPayloadVerified` / `IsPayloadVerified` /
`RequeuePendingELPayload` on `ForkChoiceStore`
- Change local-EL drain path in `chainTipSync` from `InsertBlocks` to
per-payload `NewPayload`: VALID -> mark verified, SYNCING/ACCEPTED ->
re-queue
- Add a bounded verification sweep at chain tip for unverified blocks
between finalized and head
- Gate local-EL retry/sweep machinery by `SupportInsertion()` so
standalone Caplin with a remote EL is unaffected
- Add stages coverage for anchor envelope validation, builder/self-build
signatures, pending payload retry helpers, and standalone gating

Closes erigontech#21131

## Test plan

- [x] `go test ./cl/phase1/forkchoice/... -count=1`
- [x] `go test ./cl/phase1/stages -count=1`
- [x] `go test ./cl/phase1/network/services -count=1`
- [x] `go test -tags=spectest ./cl/spectest -run '^$' -count=1`
- [x] `make lint`
- [x] `make erigon integration`

Author: domiwei

cl/phase1/forkchoice/forkchoice.go

@@ -107,6 +107,7 @@ type ForkChoiceStore struct {
 	// [New in Gloas:EIP7732] Track execution payload validation status by execution block hash.
 	// Used to check if parent execution payload has been validated/invalidated for gossip validation.
 	executionPayloadStatus *lru.Cache[common.Hash, execution_client.PayloadStatus]
+	payloadStatusByRoot    *lru.Cache[common.Hash, execution_client.PayloadStatus]
 	// [New in Gloas:EIP7732] Track execution payload gas_limit by execution block hash.
 	// Used for the is_gas_limit_target_compatible IGNORE check in bid gossip validation.
 	executionPayloadGasLimit *lru.Cache[common.Hash, uint64]
@@ -316,6 +317,10 @@ func NewForkChoiceStore(
 	if err != nil {
 		return nil, err
 	}
+	payloadStatusByRoot, err := lru.New[common.Hash, execution_client.PayloadStatus](checkpointsPerCache)
+	if err != nil {
+		return nil, err
+	}
 
 	// [New in Gloas:EIP7732] Track execution payload gas_limit by execution block hash
 	executionPayloadGasLimit, err := lru.New[common.Hash, uint64](checkpointsPerCache)
@@ -377,6 +382,7 @@ func NewForkChoiceStore(
 		pendingEnvelopes:               pendingEnvelopes,
 		pendingLocalSelfBuildEnvelopes: pendingLocalSelfBuildEnvelopes,
 		executionPayloadStatus:         executionPayloadStatus,
+		payloadStatusByRoot:            payloadStatusByRoot,
 		executionPayloadGasLimit:       executionPayloadGasLimit,
 		db:                             db,
 	}
@@ -427,6 +433,13 @@ func (f *ForkChoiceStore) GetRecentExecutionPayloadStatus(executionBlockHash com
 	return f.executionPayloadStatus.Get(executionBlockHash)
 }
 
+func (f *ForkChoiceStore) GetRecentExecutionPayloadStatusByRoot(blockRoot common.Hash) (execution_client.PayloadStatus, bool) {
+	if f.payloadStatusByRoot == nil {
+		return execution_client.PayloadStatusNone, false
+	}
+	return f.payloadStatusByRoot.Get(blockRoot)
+}
+
 // GetExecutionPayloadGasLimit returns the gas_limit of a recently validated execution payload.
 func (f *ForkChoiceStore) GetExecutionPayloadGasLimit(executionBlockHash common.Hash) (uint64, bool) {
 	return f.executionPayloadGasLimit.Get(executionBlockHash)
@@ -761,6 +774,58 @@ func (f *ForkChoiceStore) HasEnvelope(blockRoot common.Hash) bool {
 	return f.forkGraph.HasEnvelope(blockRoot)
 }
 
+// IsPayloadVerified returns whether the execution payload for the beacon block root
+// has been accepted by the execution layer.
+// [New in Gloas:EIP7732]
+func (f *ForkChoiceStore) IsPayloadVerified(blockRoot common.Hash) bool {
+	if f.verifiedExecutionPayload == nil {
+		return false
+	}
+	return f.verifiedExecutionPayload.Contains(blockRoot)
+}
+
+func (f *ForkChoiceStore) MarkPayloadVerified(blockRoot common.Hash, executionBlockHash common.Hash) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.markPayloadVerifiedLocked(blockRoot, executionBlockHash)
+}
+
+func (f *ForkChoiceStore) markPayloadVerifiedLocked(blockRoot common.Hash, executionBlockHash common.Hash) {
+	if f.verifiedExecutionPayload == nil {
+		return
+	}
+	f.verifiedExecutionPayload.Add(blockRoot, struct{}{})
+	if f.executionPayloadStatus != nil {
+		f.executionPayloadStatus.Add(executionBlockHash, execution_client.PayloadStatusValidated)
+	}
+	if f.payloadStatusByRoot != nil {
+		f.payloadStatusByRoot.Add(blockRoot, execution_client.PayloadStatusValidated)
+	}
+	f.headHash = common.Hash{}
+	f.headPayloadStatus = cltypes.PayloadStatusPending
+}
+
+func (f *ForkChoiceStore) MarkPayloadInvalid(blockRoot common.Hash, executionBlockHash common.Hash) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.markPayloadInvalidLocked(blockRoot, executionBlockHash)
+}
+
+func (f *ForkChoiceStore) markPayloadInvalidLocked(blockRoot common.Hash, executionBlockHash common.Hash) {
+	if f.verifiedExecutionPayload != nil {
+		f.verifiedExecutionPayload.Remove(blockRoot)
+	}
+	if f.executionPayloadStatus != nil {
+		f.executionPayloadStatus.Add(executionBlockHash, execution_client.PayloadStatusInvalidated)
+	}
+	if f.payloadStatusByRoot != nil {
+		f.payloadStatusByRoot.Add(blockRoot, execution_client.PayloadStatusInvalidated)
+	}
+	f.forkGraph.MarkHeaderAsInvalid(blockRoot)
+	f.headHash = common.Hash{}
+	f.headPayloadStatus = cltypes.PayloadStatusPending
+}
+
 // ReadEnvelopeFromDisk delegates to forkGraph.ReadEnvelopeFromDisk.
 // [New in Gloas:EIP7732]
 func (f *ForkChoiceStore) ReadEnvelopeFromDisk(blockRoot common.Hash) (*cltypes.SignedExecutionPayloadEnvelope, error) {
@@ -985,6 +1050,14 @@ func (f *ForkChoiceStore) GetProposerLookahead(slot uint64) (solid.Uint64VectorS
 func (f *ForkChoiceStore) addPendingELPayload(block *cltypes.SignedBeaconBlock, envelope *cltypes.SignedExecutionPayloadEnvelope) {
 	f.pendingELPayloadsMu.Lock()
 	defer f.pendingELPayloadsMu.Unlock()
+	root, ok := pendingELPayloadRoot(PendingELPayload{Block: block, Envelope: envelope})
+	if ok {
+		for _, p := range f.pendingELPayloads {
+			if existingRoot, existingOk := pendingELPayloadRoot(p); existingOk && existingRoot == root {
+				return
+			}
+		}
+	}
 	if len(f.pendingELPayloads) >= maxPendingELPayloads {
 		log.Warn("addPendingELPayload: dropping oldest pending EL payload", "queueLen", len(f.pendingELPayloads))
 		copy(f.pendingELPayloads, f.pendingELPayloads[1:])
@@ -997,8 +1070,21 @@ func (f *ForkChoiceStore) addPendingELPayload(block *cltypes.SignedBeaconBlock,
 	})
 }
 
+func pendingELPayloadRoot(p PendingELPayload) (common.Hash, bool) {
+	if p.Envelope != nil && p.Envelope.Message != nil && p.Envelope.Message.BeaconBlockRoot != (common.Hash{}) {
+		return p.Envelope.Message.BeaconBlockRoot, true
+	}
+	return common.Hash{}, false
+}
+
+// RequeuePendingELPayload queues a drained execution payload for another EL validation attempt.
+// [New in Gloas:EIP7732]
+func (f *ForkChoiceStore) RequeuePendingELPayload(p PendingELPayload) {
+	f.addPendingELPayload(p.Block, p.Envelope)
+}
+
 // DrainPendingELPayloads returns and clears all queued EL payloads.
-// The stages layer calls this before Flush() to add them to blockCollector.
+// The stages layer calls this before Flush() to retry them with engine.NewPayload.
 func (f *ForkChoiceStore) DrainPendingELPayloads() []PendingELPayload {
 	f.pendingELPayloadsMu.Lock()
 	defer f.pendingELPayloadsMu.Unlock()

cl/phase1/forkchoice/interface.go

@@ -75,8 +75,11 @@ type ForkChoiceStorageReader interface {
 	GetBlock(blockRoot common.Hash) (*cltypes.SignedBeaconBlock, bool)
 	// [New in Gloas:EIP7732] HasEnvelope checks if a signed execution payload envelope exists.
 	HasEnvelope(blockRoot common.Hash) bool
+	// [New in Gloas:EIP7732] IsPayloadVerified checks whether the execution payload was accepted by the EL.
+	IsPayloadVerified(blockRoot common.Hash) bool
 	// [New in Gloas:EIP7732] ReadEnvelopeFromDisk reads a signed execution payload envelope from disk.
 	ReadEnvelopeFromDisk(blockRoot common.Hash) (*cltypes.SignedExecutionPayloadEnvelope, error)
+	GetRecentExecutionPayloadStatusByRoot(blockRoot common.Hash) (execution_client.PayloadStatus, bool)
 	// [New in Gloas:EIP7732] IsBlobDataAvailable returns the local node's assessment of whether
 	// blob data is available for the given block. Used by the payload_attestation_data API so PTC
 	// validators can set the blob_data_available flag independently of payload_present.

cl/phase1/forkchoice/mock_services/forkchoice_mock.go

@@ -69,6 +69,7 @@ type ForkChoiceStorageMock struct {
 	Headers                   map[common.Hash]*cltypes.BeaconBlockHeader
 	Blocks                    map[common.Hash]*cltypes.SignedBeaconBlock
 	Envelopes                 map[common.Hash]*cltypes.SignedExecutionPayloadEnvelope
+	VerifiedPayloads          map[common.Hash]bool
 	GetBeaconCommitteeMock    func(slot, committeeIndex uint64) ([]uint64, error)
 
 	Pool pool.OperationsPool
@@ -82,6 +83,7 @@ type ForkChoiceStorageMock struct {
 
 	// [New in Gloas:EIP7732] Execution payload status by execution block hash
 	ExecutionPayloadStatusMap map[common.Hash]execution_client.PayloadStatus
+	PayloadStatusByRootMap    map[common.Hash]execution_client.PayloadStatus
 	// [New in Gloas:EIP7732] Execution payload gas limit by execution block hash
 	ExecutionPayloadGasLimitMap map[common.Hash]uint64
 }
@@ -213,6 +215,7 @@ func NewForkChoiceStorageMock(t *testing.T) *ForkChoiceStorageMock {
 		SyncContributionPool:        makeSyncContributionPoolMock(t),
 		MockPeerDas:                 mockPeerDas,
 		ExecutionPayloadStatusMap:   make(map[common.Hash]execution_client.PayloadStatus),
+		PayloadStatusByRootMap:      make(map[common.Hash]execution_client.PayloadStatus),
 		ExecutionPayloadGasLimitMap: make(map[common.Hash]uint64),
 	}
 }
@@ -426,6 +429,13 @@ func (f *ForkChoiceStorageMock) HasEnvelope(blockRoot common.Hash) bool {
 	return ok
 }
 
+func (f *ForkChoiceStorageMock) IsPayloadVerified(blockRoot common.Hash) bool {
+	if f.VerifiedPayloads == nil {
+		return false
+	}
+	return f.VerifiedPayloads[blockRoot]
+}
+
 func (f *ForkChoiceStorageMock) ReadEnvelopeFromDisk(blockRoot common.Hash) (*cltypes.SignedExecutionPayloadEnvelope, error) {
 	return f.Envelopes[blockRoot], nil
 }
@@ -530,6 +540,11 @@ func (f *ForkChoiceStorageMock) GetRecentExecutionPayloadStatus(executionBlockHa
 	return status, ok
 }
 
+func (f *ForkChoiceStorageMock) GetRecentExecutionPayloadStatusByRoot(blockRoot common.Hash) (execution_client.PayloadStatus, bool) {
+	status, ok := f.PayloadStatusByRootMap[blockRoot]
+	return status, ok
+}
+
 // GetExecutionPayloadGasLimit returns the gas_limit of a recently validated execution payload.
 // [New in Gloas:EIP7732]
 func (f *ForkChoiceStorageMock) GetExecutionPayloadGasLimit(executionBlockHash common.Hash) (uint64, bool) {

cl/phase1/forkchoice/on_attestation.go

@@ -282,7 +282,7 @@ func (f *ForkChoiceStore) ValidateOnAttestation(attestation *solid.Attestation)
 			return errors.New("attestation index must be 0 when block_slot equals attestation_slot")
 		}
 		// PTC attestation (index 1): payload must be verified
-		if attestation.Data.CommitteeIndex == 1 && !f.forkGraph.HasEnvelope(attestation.Data.BeaconBlockRoot) {
+		if attestation.Data.CommitteeIndex == 1 && !f.IsPayloadVerified(attestation.Data.BeaconBlockRoot) {
 			return errors.New("PTC attestation requires verified payload envelope")
 		}
 	}

cl/phase1/forkchoice/on_execution_payload.go

@@ -296,19 +296,11 @@ func (f *ForkChoiceStore) validatePayloadWithEL(
 		}
 	case execution_client.PayloadStatusInvalidated:
 		log.Warn("validatePayloadWithEL: payload is invalid", "beaconBlockRoot", beaconBlockRoot, "err", err)
-		f.forkGraph.MarkHeaderAsInvalid(beaconBlockRoot)
-		// remove from optimistic candidate
-		if err := f.optimisticStore.InvalidateBlock(beaconBlockRoot, block.Block); err != nil {
-			return fmt.Errorf("failed to remove block from optimistic store: %v", err)
-		}
+		f.markPayloadInvalidLocked(beaconBlockRoot, executionBlockHash)
 		return errors.New("execution payload is invalid")
 	case execution_client.PayloadStatusValidated:
 		log.Trace("validatePayloadWithEL: payload is validated", "beaconBlockRoot", beaconBlockRoot)
-		// remove from optimistic candidate
-		if err := f.optimisticStore.ValidateBlock(beaconBlockRoot, block.Block); err != nil {
-			return fmt.Errorf("failed to validate block in optimistic store: %v", err)
-		}
-		f.verifiedExecutionPayload.Add(beaconBlockRoot, struct{}{})
+		f.markPayloadVerifiedLocked(beaconBlockRoot, executionBlockHash)
 	}
 
 	if err != nil {
@@ -451,24 +443,24 @@ func (f *ForkChoiceStore) applyEnvelopeLocked(ctx context.Context, signedEnvelop
 // on disk to resolve parent execution payloads for subsequent blocks.
 // [New in Gloas:EIP7732]
 func (f *ForkChoiceStore) StoreAnchorEnvelope(blockRoot common.Hash, signedEnvelope *cltypes.SignedExecutionPayloadEnvelope) error {
-	if signedEnvelope == nil || signedEnvelope.Message == nil {
+	if signedEnvelope == nil || signedEnvelope.Message == nil || signedEnvelope.Message.Payload == nil {
 		return errors.New("StoreAnchorEnvelope: nil envelope")
 	}
 	envelope := signedEnvelope.Message
+	if envelope.BeaconBlockRoot != blockRoot {
+		return fmt.Errorf("StoreAnchorEnvelope: envelope root %v does not match block root %v", envelope.BeaconBlockRoot, blockRoot)
+	}
 
 	f.mu.Lock()
-	// Update eth2Roots mapping so FCU can resolve the EL block hash
-	if envelope.Payload != nil {
-		f.eth2Roots.Add(blockRoot, envelope.Payload.BlockHash)
-	}
-	// Persist to disk so HasEnvelope() returns true and forward sync can find it
 	if err := f.forkGraph.DumpEnvelopeOnDisk(blockRoot, signedEnvelope); err != nil {
 		f.mu.Unlock()
 		return fmt.Errorf("StoreAnchorEnvelope: failed to dump envelope: %w", err)
 	}
+	f.eth2Roots.Add(blockRoot, envelope.Payload.BlockHash)
+	f.headHash = common.Hash{}
+	f.headPayloadStatus = cltypes.PayloadStatusPending
 	f.mu.Unlock()
 
-	// Write DB indices outside the lock
 	if f.db != nil {
 		ctx := context.Background()
 		if err := f.db.Update(ctx, func(tx kv.RwTx) error {
@@ -477,6 +469,7 @@ func (f *ForkChoiceStore) StoreAnchorEnvelope(blockRoot common.Hash, signedEnvel
 			return fmt.Errorf("StoreAnchorEnvelope: failed to write indices: %w", err)
 		}
 	}
+
 	return nil
 }
 

cl/phase1/forkchoice/on_execution_payload_test.go

@@ -19,12 +19,16 @@ package forkchoice
 import (
 	"context"
 	"testing"
+	"time"
 
+	"github.com/hashicorp/golang-lru/v2"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 
 	"github.com/erigontech/erigon/cl/clparams"
 	"github.com/erigontech/erigon/cl/cltypes"
 	"github.com/erigontech/erigon/cl/cltypes/solid"
+	"github.com/erigontech/erigon/cl/phase1/execution_client"
 	"github.com/erigontech/erigon/common"
 )
 
@@ -271,3 +275,90 @@ func TestValidatePayloadWithEL_NoEngine(t *testing.T) {
 	err := f.validatePayloadWithEL(context.TODO(), envelope, block, common.Hash{})
 	require.NoError(t, err)
 }
+
+func TestValidatePayloadWithELDoesNotRelockForkChoiceMu(t *testing.T) {
+	cfg := &clparams.MainnetBeaconConfig
+	for _, tt := range []struct {
+		name       string
+		status     execution_client.PayloadStatus
+		wantErr    bool
+		wantVerify bool
+	}{
+		{
+			name:       "validated",
+			status:     execution_client.PayloadStatusValidated,
+			wantVerify: true,
+		},
+		{
+			name:    "invalidated",
+			status:  execution_client.PayloadStatusInvalidated,
+			wantErr: true,
+		},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			engine := execution_client.NewMockExecutionEngine(ctrl)
+			engine.EXPECT().
+				NewPayload(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+				Return(tt.status, nil)
+
+			verifiedExecutionPayload, err := lru.New[common.Hash, struct{}](16)
+			require.NoError(t, err)
+			executionPayloadStatus, err := lru.New[common.Hash, execution_client.PayloadStatus](16)
+			require.NoError(t, err)
+			payloadStatusByRoot, err := lru.New[common.Hash, execution_client.PayloadStatus](16)
+			require.NoError(t, err)
+			executionPayloadGasLimit, err := lru.New[common.Hash, uint64](16)
+			require.NoError(t, err)
+
+			blockRoot := common.HexToHash("0x1234")
+			executionBlockHash := common.HexToHash("0xabcd")
+			invalidatedHeader := common.Hash{}
+			f := &ForkChoiceStore{
+				beaconCfg:                cfg,
+				engine:                   engine,
+				forkGraph:                payloadVoteForkGraph{invalidatedHeader: &invalidatedHeader},
+				verifiedExecutionPayload: verifiedExecutionPayload,
+				executionPayloadStatus:   executionPayloadStatus,
+				payloadStatusByRoot:      payloadStatusByRoot,
+				executionPayloadGasLimit: executionPayloadGasLimit,
+			}
+			envelope := &cltypes.ExecutionPayloadEnvelope{
+				Payload: &cltypes.Eth1Block{BlockHash: executionBlockHash},
+			}
+			body := cltypes.NewBeaconBody(cfg, clparams.GloasVersion)
+			body.SignedExecutionPayloadBid = &cltypes.SignedExecutionPayloadBid{
+				Message: &cltypes.ExecutionPayloadBid{
+					BlobKzgCommitments: *solid.NewStaticListSSZ[*cltypes.KZGCommitment](0, 48),
+				},
+			}
+			block := &cltypes.SignedBeaconBlock{
+				Block: &cltypes.BeaconBlock{
+					Body: body,
+				},
+			}
+
+			done := make(chan error, 1)
+			go func() {
+				f.mu.Lock()
+				defer f.mu.Unlock()
+				done <- f.validatePayloadWithEL(context.Background(), envelope, block, blockRoot)
+			}()
+
+			select {
+			case err := <-done:
+				if tt.wantErr {
+					require.Error(t, err)
+				} else {
+					require.NoError(t, err)
+				}
+			case <-time.After(time.Second):
+				t.Fatal("validatePayloadWithEL blocked while forkchoice mutex was already held")
+			}
+			require.Equal(t, tt.wantVerify, f.IsPayloadVerified(blockRoot))
+			if tt.status == execution_client.PayloadStatusInvalidated {
+				require.Equal(t, blockRoot, invalidatedHeader)
+			}
+		})
+	}
+}