diff --git a/cl/beacon/handler/epbs.go b/cl/beacon/handler/epbs.go index 84422149945..473e5fd75b9 100644 --- a/cl/beacon/handler/epbs.go +++ b/cl/beacon/handler/epbs.go @@ -816,7 +816,7 @@ func (a *ApiHandler) PostEthV1BeaconExecutionPayloadEnvelope(w http.ResponseWrit // checkBlobData=false because gossip validation handles it; validatePayload=true // so the EL receives NewPayload for the execution payload. if err := a.forkchoiceStore.OnExecutionPayload(r.Context(), signedEnvelope, false, true); err != nil { - if errors.Is(err, forkchoice.ErrIgnore) || errors.Is(err, forkchoice.ErrEIP7594ColumnDataNotAvailable) { + if errors.Is(err, forkchoice.ErrIgnore) || errors.Is(err, forkchoice.ErrExecutionPayloadAlreadyStored) || errors.Is(err, forkchoice.ErrEIP7594ColumnDataNotAvailable) { a.logger.Debug("[Beacon REST] OnExecutionPayload queued or ignored", "err", err) } else { beaconhttp.WrapEndpointError(err).WriteTo(w) diff --git a/cl/beacon/handler/epbs_test.go b/cl/beacon/handler/epbs_test.go index b2ad7acc422..315f27cbfc4 100644 --- a/cl/beacon/handler/epbs_test.go +++ b/cl/beacon/handler/epbs_test.go @@ -33,6 +33,7 @@ import ( "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/forkchoice" "github.com/erigontech/erigon/cl/phase1/network/services" mock_services "github.com/erigontech/erigon/cl/phase1/network/services/mock_services" "github.com/erigontech/erigon/cl/pool" @@ -165,6 +166,19 @@ func TestPostExecutionPayloadEnvelopeReturnsForkchoiceError(t *testing.T) { require.Contains(t, recorder.Body.String(), "invalid execution payload") } +func TestPostExecutionPayloadEnvelopeAcceptsAlreadyStored(t *testing.T) { + _, _, _, _, _, handler, _, _, fcu, _ := setupTestingHandler(t, clparams.BellatrixVersion, log.Root(), true) + fcu.OnExecutionPayloadErr = forkchoice.ErrExecutionPayloadAlreadyStored + + request := httptest.NewRequest(http.MethodPost, "/eth/v1/beacon/execution_payload_envelope", strings.NewReader(`{}`)) + request.Header.Set("Content-Type", "application/json; charset=utf-8") + recorder := httptest.NewRecorder() + + handler.PostEthV1BeaconExecutionPayloadEnvelope(recorder, request) + + require.Equal(t, http.StatusOK, recorder.Code, recorder.Body.String()) +} + func TestPostPtcDutiesDoesNotCapValidatorCount(t *testing.T) { _, _, _, _, _, handler, _, _, _, _ := setupTestingHandler(t, clparams.BellatrixVersion, log.Root(), true) handler.beaconChainCfg.GloasForkEpoch = 0 diff --git a/cl/phase1/forkchoice/fork_graph/fork_graph_disk_fs.go b/cl/phase1/forkchoice/fork_graph/fork_graph_disk_fs.go index f8b728db142..23ebd3e34a2 100644 --- a/cl/phase1/forkchoice/fork_graph/fork_graph_disk_fs.go +++ b/cl/phase1/forkchoice/fork_graph/fork_graph_disk_fs.go @@ -18,6 +18,7 @@ package fork_graph import ( "encoding/binary" + "errors" "fmt" "io" "os" @@ -193,6 +194,11 @@ func (f *forkGraphDisk) HasEnvelope(blockRoot common.Hash) bool { if _, ok := f.envelopeExists.Load(blockRoot); ok { return true } + f.stateDumpLock.Lock() + defer f.stateDumpLock.Unlock() + if _, ok := f.envelopeExists.Load(blockRoot); ok { + return true + } // Slow path: fall back to disk and populate cache on hit exists, err := afero.Exists(f.fs, getEnvelopeFilename(blockRoot)) if err == nil && exists { @@ -206,19 +212,35 @@ func (f *forkGraphDisk) HasEnvelope(blockRoot common.Hash) bool { // [New in Gloas:EIP7732] func (f *forkGraphDisk) ReadEnvelopeFromDisk(blockRoot common.Hash) (envelope *cltypes.SignedExecutionPayloadEnvelope, err error) { var file afero.File + var corrupt bool f.stateDumpLock.Lock() defer f.stateDumpLock.Unlock() - file, err = f.fs.Open(getEnvelopeFilename(blockRoot)) + filename := getEnvelopeFilename(blockRoot) + file, err = f.fs.Open(filename) if err != nil { + f.envelopeExists.Delete(blockRoot) return } - defer file.Close() + defer func() { + if closeErr := file.Close(); closeErr != nil { + log.Warn("failed to close envelope after read", "root", blockRoot, "err", closeErr) + } + if corrupt { + if removeErr := f.fs.Remove(filename); removeErr != nil && !os.IsNotExist(removeErr) { + log.Warn("failed to remove corrupt envelope", "root", blockRoot, "err", removeErr) + } + } + if err != nil { + f.envelopeExists.Delete(blockRoot) + } + }() + readTracker := &envelopeReadTracker{Reader: file} if f.sszSnappyReader == nil { - f.sszSnappyReader = snappy.NewReader(file) + f.sszSnappyReader = snappy.NewReader(readTracker) } else { - f.sszSnappyReader.Reset(file) + f.sszSnappyReader.Reset(readTracker) } // Read the length @@ -226,37 +248,55 @@ func (f *forkGraphDisk) ReadEnvelopeFromDisk(blockRoot common.Hash) (envelope *c var n int n, err = io.ReadFull(f.sszSnappyReader, lengthBytes) if err != nil { + corrupt = isCorruptEnvelopeReadError(err, readTracker.err) return nil, fmt.Errorf("failed to read length: %w, root: %x", err, blockRoot) } if n != 8 { + corrupt = true return nil, fmt.Errorf("failed to read length: %d, want 8, root: %x", n, blockRoot) } envelopeLength := binary.BigEndian.Uint64(lengthBytes) if envelopeLength > maxSSZObjectSize { + corrupt = true return nil, fmt.Errorf("corrupt envelope file: length %d exceeds max %d, root: %x", envelopeLength, maxSSZObjectSize, blockRoot) } - if envelopeLength > uint64(cap(f.sszBuffer)) { - f.sszBuffer = make([]byte, envelopeLength) - } else { - f.sszBuffer = f.sszBuffer[:envelopeLength] - } - n, err = io.ReadFull(f.sszSnappyReader, f.sszBuffer) + ownedBuffer := make([]byte, envelopeLength) + n, err = io.ReadFull(f.sszSnappyReader, ownedBuffer) if err != nil { + corrupt = isCorruptEnvelopeReadError(err, readTracker.err) return nil, fmt.Errorf("failed to read snappy buffer: %w, root: %x", err, blockRoot) } - f.sszBuffer = f.sszBuffer[:n] + ownedBuffer = ownedBuffer[:n] envelope = &cltypes.SignedExecutionPayloadEnvelope{ Message: cltypes.NewExecutionPayloadEnvelope(f.beaconCfg), } - if err = envelope.DecodeSSZ(f.sszBuffer, int(clparams.GloasVersion)); err != nil { + if err = envelope.DecodeSSZ(ownedBuffer, int(clparams.GloasVersion)); err != nil { + corrupt = true return nil, fmt.Errorf("failed to decode envelope: %w, root: %x, len: %d", err, blockRoot, n) } return } +type envelopeReadTracker struct { + io.Reader + err error +} + +func (r *envelopeReadTracker) Read(p []byte) (int, error) { + n, err := r.Reader.Read(p) + if err != nil && !errors.Is(err, io.EOF) { + r.err = err + } + return n, err +} + +func isCorruptEnvelopeReadError(err, sourceErr error) bool { + return sourceErr == nil || !errors.Is(err, sourceErr) +} + // DumpEnvelopeOnDisk dumps an execution payload envelope to disk. // [New in Gloas:EIP7732] func (f *forkGraphDisk) DumpEnvelopeOnDisk(blockRoot common.Hash, envelope *cltypes.SignedExecutionPayloadEnvelope) (err error) { @@ -276,11 +316,21 @@ func (f *forkGraphDisk) DumpEnvelopeOnDisk(blockRoot common.Hash, envelope *clty return } - dumpedFile, err := f.fs.OpenFile(getEnvelopeFilename(blockRoot), os.O_TRUNC|os.O_CREATE|os.O_RDWR, 0o755) + filename := getEnvelopeFilename(blockRoot) + tempFilename := filename + ".tmp" + dumpedFile, err := f.fs.OpenFile(tempFilename, os.O_TRUNC|os.O_CREATE|os.O_RDWR, 0o644) if err != nil { return err } - defer dumpedFile.Close() + closed := false + defer func() { + if !closed { + _ = dumpedFile.Close() + } + if err != nil { + _ = f.fs.Remove(tempFilename) + } + }() if f.sszSnappyWriter == nil { f.sszSnappyWriter = snappy.NewBufferedWriter(dumpedFile) @@ -309,6 +359,13 @@ func (f *forkGraphDisk) DumpEnvelopeOnDisk(blockRoot common.Hash, envelope *clty log.Error("failed to sync dumped file", "err", err) return } + if err = dumpedFile.Close(); err != nil { + return + } + closed = true + if err = f.fs.Rename(tempFilename, filename); err != nil { + return + } return } diff --git a/cl/phase1/forkchoice/fork_graph/fork_graph_test.go b/cl/phase1/forkchoice/fork_graph/fork_graph_test.go index 55cbbd11d83..a4a3f18a03c 100644 --- a/cl/phase1/forkchoice/fork_graph/fork_graph_test.go +++ b/cl/phase1/forkchoice/fork_graph/fork_graph_test.go @@ -18,14 +18,20 @@ package fork_graph import ( _ "embed" + "errors" + "os" + "strings" + "sync" "testing" "github.com/erigontech/erigon/cl/beacon/beacon_router_configuration" "github.com/erigontech/erigon/cl/phase1/core/state" + "github.com/golang/snappy" "github.com/spf13/afero" "github.com/erigontech/erigon/cl/clparams" "github.com/erigontech/erigon/cl/cltypes" + "github.com/erigontech/erigon/cl/cltypes/solid" "github.com/erigontech/erigon/cl/utils" "github.com/erigontech/erigon/common" "github.com/stretchr/testify/require" @@ -40,6 +46,99 @@ var block2 []byte //go:embed test_data/anchor_state.ssz_snappy var anchor []byte +var errTestEnvelopeIO = errors.New("test envelope I/O error") + +type envelopeCloseErrorFile struct { + afero.File +} + +func (f envelopeCloseErrorFile) Close() error { + _ = f.File.Close() + return errTestEnvelopeIO +} + +type envelopeCloseErrorFs struct { + afero.Fs +} + +func (f envelopeCloseErrorFs) Open(name string) (afero.File, error) { + file, err := f.Fs.Open(name) + if err != nil { + return nil, err + } + return envelopeCloseErrorFile{File: file}, nil +} + +type envelopeReadErrorFile struct { + afero.File +} + +func (envelopeReadErrorFile) Read([]byte) (int, error) { + return 0, errTestEnvelopeIO +} + +type envelopeReadErrorFs struct { + afero.Fs +} + +type envelopeWriteFailureFs struct { + afero.Fs + stage string +} + +func (f envelopeWriteFailureFs) OpenFile(name string, flag int, perm os.FileMode) (afero.File, error) { + if f.stage == "open" && strings.HasSuffix(name, ".tmp") { + return nil, errTestEnvelopeIO + } + file, err := f.Fs.OpenFile(name, flag, perm) + if err != nil { + return nil, err + } + return envelopeWriteFailureFile{File: file, stage: f.stage}, nil +} + +func (f envelopeWriteFailureFs) Rename(oldname, newname string) error { + if f.stage == "rename" { + return errTestEnvelopeIO + } + return f.Fs.Rename(oldname, newname) +} + +type envelopeWriteFailureFile struct { + afero.File + stage string +} + +func (f envelopeWriteFailureFile) Write(p []byte) (int, error) { + if f.stage == "write" { + return 0, errTestEnvelopeIO + } + return f.File.Write(p) +} + +func (f envelopeWriteFailureFile) Sync() error { + if f.stage == "sync" { + return errTestEnvelopeIO + } + return f.File.Sync() +} + +func (f envelopeWriteFailureFile) Close() error { + if f.stage == "close" { + _ = f.File.Close() + return errTestEnvelopeIO + } + return f.File.Close() +} + +func (f envelopeReadErrorFs) Open(name string) (afero.File, error) { + file, err := f.Fs.Open(name) + if err != nil { + return nil, err + } + return envelopeReadErrorFile{File: file}, nil +} + func TestForkGraphInDisk(t *testing.T) { blockA, blockB, blockC := cltypes.NewSignedBeaconBlock(&clparams.MainnetBeaconConfig, clparams.DenebVersion), cltypes.NewSignedBeaconBlock(&clparams.MainnetBeaconConfig, clparams.DenebVersion), @@ -87,3 +186,175 @@ func TestPruneKeepsLowestAvailableBlockMonotonic(t *testing.T) { require.NoError(t, f.Prune(120)) require.Equal(t, uint64(151), f.LowestAvailableSlot()) } + +func TestReadEnvelopeRemovesCorruptPersistenceMarker(t *testing.T) { + fs := afero.NewMemMapFs() + f := &forkGraphDisk{fs: fs, beaconCfg: &clparams.MainnetBeaconConfig} + root := common.HexToHash("0x1234") + require.NoError(t, afero.WriteFile(fs, getEnvelopeFilename(root), []byte("truncated"), 0o644)) + require.True(t, f.HasEnvelope(root)) + + _, err := f.ReadEnvelopeFromDisk(root) + require.Error(t, err) + require.False(t, f.HasEnvelope(root)) +} + +func TestReadEnvelopeRemovesUnsupportedSnappyFrames(t *testing.T) { + streamIdentifier := []byte{0xff, 0x06, 0x00, 0x00, 's', 'N', 'a', 'P', 'p', 'Y'} + for _, tt := range []struct { + name string + frame []byte + wantErr error + }{ + { + name: "reserved unskippable chunk", + frame: append(append([]byte{}, streamIdentifier...), 0x02, 0x00, 0x00, 0x00), + wantErr: snappy.ErrUnsupported, + }, + } { + t.Run(tt.name, func(t *testing.T) { + fs := afero.NewMemMapFs() + f := &forkGraphDisk{fs: fs, beaconCfg: &clparams.MainnetBeaconConfig} + root := common.HexToHash("0x1234") + require.NoError(t, afero.WriteFile(fs, getEnvelopeFilename(root), tt.frame, 0o644)) + require.True(t, f.HasEnvelope(root)) + + _, err := f.ReadEnvelopeFromDisk(root) + require.ErrorIs(t, err, tt.wantErr) + require.False(t, f.HasEnvelope(root)) + }) + } +} + +func TestEnvelopeReadClassifiesSnappyStructuralErrors(t *testing.T) { + require.True(t, isCorruptEnvelopeReadError(snappy.ErrCorrupt, nil)) + require.True(t, isCorruptEnvelopeReadError(snappy.ErrUnsupported, nil)) + require.True(t, isCorruptEnvelopeReadError(snappy.ErrTooLarge, nil)) + require.False(t, isCorruptEnvelopeReadError(errTestEnvelopeIO, errTestEnvelopeIO)) +} + +func TestDumpEnvelopeAtomicallyPersistsReadableFile(t *testing.T) { + fs := afero.NewMemMapFs() + f := &forkGraphDisk{fs: fs, beaconCfg: &clparams.MainnetBeaconConfig} + root := common.HexToHash("0x1234") + envelope := cltypes.NewExecutionPayloadEnvelope(&clparams.MainnetBeaconConfig) + envelope.BeaconBlockRoot = root + envelope.Payload.Extra = solid.NewExtraData() + envelope.Payload.Transactions = &solid.TransactionsSSZ{} + + require.NoError(t, f.DumpEnvelopeOnDisk(root, &cltypes.SignedExecutionPayloadEnvelope{Message: envelope})) + tempExists, err := afero.Exists(f.fs, getEnvelopeFilename(root)+".tmp") + require.NoError(t, err) + require.False(t, tempExists) + persisted, err := f.ReadEnvelopeFromDisk(root) + require.NoError(t, err) + require.Equal(t, root, persisted.Message.BeaconBlockRoot) +} + +func TestDumpEnvelopeFailurePreservesExistingFinal(t *testing.T) { + for _, stage := range []string{"open", "write", "sync", "close", "rename"} { + t.Run(stage, func(t *testing.T) { + fs := afero.NewMemMapFs() + f := &forkGraphDisk{fs: fs, beaconCfg: &clparams.MainnetBeaconConfig} + root := common.HexToHash("0x1234") + require.NoError(t, f.DumpEnvelopeOnDisk(root, testEnvelopeWithTransaction(root, []byte{1, 2, 3}))) + + f.fs = envelopeWriteFailureFs{Fs: fs, stage: stage} + require.ErrorIs(t, f.DumpEnvelopeOnDisk(root, testEnvelopeWithTransaction(root, []byte{9, 8, 7})), errTestEnvelopeIO) + f.fs = fs + + tempExists, err := afero.Exists(fs, getEnvelopeFilename(root)+".tmp") + require.NoError(t, err) + require.False(t, tempExists) + persisted, err := f.ReadEnvelopeFromDisk(root) + require.NoError(t, err) + require.Equal(t, [][]byte{{1, 2, 3}}, persisted.Message.Payload.Transactions.UnderlyngReference()) + }) + } +} + +func TestReadEnvelopeOwnsDecodedTransactions(t *testing.T) { + fs := afero.NewMemMapFs() + f := &forkGraphDisk{fs: fs, beaconCfg: &clparams.MainnetBeaconConfig} + rootA := common.HexToHash("0xa") + rootB := common.HexToHash("0xb") + envelopeA := testEnvelopeWithTransaction(rootA, []byte{1, 2, 3}) + envelopeB := testEnvelopeWithTransaction(rootB, []byte{9, 8, 7}) + require.NoError(t, f.DumpEnvelopeOnDisk(rootA, envelopeA)) + + persistedA, err := f.ReadEnvelopeFromDisk(rootA) + require.NoError(t, err) + require.NoError(t, f.DumpEnvelopeOnDisk(rootB, envelopeB)) + _, err = f.ReadEnvelopeFromDisk(rootB) + require.NoError(t, err) + require.Equal(t, [][]byte{{1, 2, 3}}, persistedA.Message.Payload.Transactions.UnderlyngReference()) +} + +func TestReadEnvelopeTransactionsDoNotRaceWithDump(t *testing.T) { + fs := afero.NewMemMapFs() + f := &forkGraphDisk{fs: fs, beaconCfg: &clparams.MainnetBeaconConfig} + rootA := common.HexToHash("0xa") + rootB := common.HexToHash("0xb") + require.NoError(t, f.DumpEnvelopeOnDisk(rootA, testEnvelopeWithTransaction(rootA, []byte{1, 2, 3}))) + persistedA, err := f.ReadEnvelopeFromDisk(rootA) + require.NoError(t, err) + envelopeB := testEnvelopeWithTransaction(rootB, []byte{9, 8, 7}) + + start := make(chan struct{}) + errCh := make(chan error, 1) + var wg sync.WaitGroup + wg.Go(func() { + <-start + for range 100 { + if err := f.DumpEnvelopeOnDisk(rootB, envelopeB); err != nil { + errCh <- err + return + } + } + }) + close(start) + var observed uint64 + for range 100 { + observed += uint64(persistedA.Message.Payload.Transactions.UnderlyngReference()[0][0]) + } + wg.Wait() + require.Equal(t, uint64(100), observed) + require.Equal(t, [][]byte{{1, 2, 3}}, persistedA.Message.Payload.Transactions.UnderlyngReference()) + close(errCh) + for err := range errCh { + require.NoError(t, err) + } +} + +func TestReadEnvelopeCloseErrorKeepsDecodedFile(t *testing.T) { + fs := afero.NewMemMapFs() + f := &forkGraphDisk{fs: fs, beaconCfg: &clparams.MainnetBeaconConfig} + root := common.HexToHash("0x1234") + require.NoError(t, f.DumpEnvelopeOnDisk(root, testEnvelopeWithTransaction(root, []byte{1}))) + f.fs = envelopeCloseErrorFs{Fs: fs} + + envelope, err := f.ReadEnvelopeFromDisk(root) + require.NoError(t, err) + require.NotNil(t, envelope) + require.True(t, f.HasEnvelope(root)) +} + +func TestReadEnvelopeTransientReadErrorKeepsFile(t *testing.T) { + fs := afero.NewMemMapFs() + f := &forkGraphDisk{fs: fs, beaconCfg: &clparams.MainnetBeaconConfig} + root := common.HexToHash("0x1234") + require.NoError(t, f.DumpEnvelopeOnDisk(root, testEnvelopeWithTransaction(root, []byte{1}))) + f.fs = envelopeReadErrorFs{Fs: fs} + + _, err := f.ReadEnvelopeFromDisk(root) + require.ErrorIs(t, err, errTestEnvelopeIO) + require.True(t, f.HasEnvelope(root)) +} + +func testEnvelopeWithTransaction(root common.Hash, transaction []byte) *cltypes.SignedExecutionPayloadEnvelope { + envelope := cltypes.NewExecutionPayloadEnvelope(&clparams.MainnetBeaconConfig) + envelope.BeaconBlockRoot = root + envelope.Payload.Extra = solid.NewExtraData() + envelope.Payload.Transactions = solid.NewTransactionsSSZFromTransactions([][]byte{transaction}) + return &cltypes.SignedExecutionPayloadEnvelope{Message: envelope} +} diff --git a/cl/phase1/forkchoice/forkchoice.go b/cl/phase1/forkchoice/forkchoice.go index 708fbae19ca..f0a745376a5 100644 --- a/cl/phase1/forkchoice/forkchoice.go +++ b/cl/phase1/forkchoice/forkchoice.go @@ -18,6 +18,7 @@ package forkchoice import ( "cmp" + "container/list" "slices" "sync" "sync/atomic" @@ -94,16 +95,19 @@ type ForkChoiceStore struct { unrealizedJustifiedCheckpoint atomic.Value unrealizedFinalizedCheckpoint atomic.Value - proposerBoostRoot atomic.Value - headHash common.Hash - headSlot uint64 - headPayloadStatus cltypes.PayloadStatus - genesisTime uint64 - genesisValidatorsRoot common.Hash - weights map[common.Hash]uint64 - headSet map[common.Hash]struct{} - hotSidecars map[common.Hash][]*cltypes.BlobSidecar // Set of sidecars that are not yet processed. - verifiedExecutionPayload *lru.Cache[common.Hash, struct{}] + proposerBoostRoot atomic.Value + headHash common.Hash + headSlot uint64 + headPayloadStatus cltypes.PayloadStatus + genesisTime uint64 + genesisValidatorsRoot common.Hash + weights map[common.Hash]uint64 + headSet map[common.Hash]struct{} + gloasVerificationLeaves *list.List + gloasVerificationLeafByRoot map[common.Hash]*list.Element + gloasVerificationLeafCursor *list.Element + hotSidecars map[common.Hash][]*cltypes.BlobSidecar // Set of sidecars that are not yet processed. + verifiedExecutionPayload *lru.Cache[common.Hash, 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] @@ -199,8 +203,9 @@ type ForkChoiceStore struct { // whose EL newPayload failed (e.g. because EL hasn't caught up after forward sync). // The stages layer drains these into blockCollector before each Flush() so EL // eventually receives the blocks. - pendingELPayloadsMu sync.Mutex - pendingELPayloads []PendingELPayload + pendingELPayloadsMu sync.Mutex + pendingELPayloads []PendingELPayload + pendingELPayloadRoots map[common.Hash]struct{} // db is used to persist execution payload indices (block number/hash) when an envelope // is accepted in OnExecutionPayload. May be nil (e.g. in tests), in which case the @@ -420,6 +425,7 @@ func NewForkChoiceStore( f.payloadDataAvailabilityVote.Store(common.Hash(anchorRoot), anchorDataAvailabilityVotes) f.gloasWeightTree = newGloasWeightTree(f) + f.initializeGloasVerificationLeaves() return f, nil } @@ -745,6 +751,71 @@ func (f *ForkChoiceStore) ForkNodes() []ForkNode { return forkNodes } +func (f *ForkChoiceStore) initializeGloasVerificationLeaves() { + f.gloasVerificationLeaves = list.New() + f.gloasVerificationLeafByRoot = make(map[common.Hash]*list.Element, len(f.headSet)) + for root := range f.headSet { + f.addGloasVerificationLeaf(root) + } +} + +func (f *ForkChoiceStore) addGloasVerificationLeaf(root common.Hash) { + if f.gloasVerificationLeaves == nil { + return + } + if _, ok := f.gloasVerificationLeafByRoot[root]; ok { + return + } + f.gloasVerificationLeafByRoot[root] = f.gloasVerificationLeaves.PushBack(root) +} + +func (f *ForkChoiceStore) removeGloasVerificationLeaf(root common.Hash) { + if f.gloasVerificationLeaves == nil { + return + } + element, ok := f.gloasVerificationLeafByRoot[root] + if !ok { + return + } + if f.gloasVerificationLeafCursor == element { + if f.gloasVerificationLeaves.Len() == 1 { + f.gloasVerificationLeafCursor = nil + } else if previous := element.Prev(); previous != nil { + f.gloasVerificationLeafCursor = previous + } else { + f.gloasVerificationLeafCursor = f.gloasVerificationLeaves.Back() + } + } + f.gloasVerificationLeaves.Remove(element) + delete(f.gloasVerificationLeafByRoot, root) +} + +// GloasVerificationLeaves returns the next bounded page of fork-tree leaves. +func (f *ForkChoiceStore) GloasVerificationLeaves(limit int) []common.Hash { + if limit <= 0 { + return nil + } + f.mu.Lock() + defer f.mu.Unlock() + if f.gloasVerificationLeaves == nil || f.gloasVerificationLeaves.Len() == 0 { + return nil + } + + leaves := make([]common.Hash, 0, min(limit, f.gloasVerificationLeaves.Len())) + for visited := 0; visited < f.gloasVerificationLeaves.Len() && len(leaves) < limit; visited++ { + if f.gloasVerificationLeafCursor == nil || f.gloasVerificationLeafCursor.Next() == nil { + f.gloasVerificationLeafCursor = f.gloasVerificationLeaves.Front() + } else { + f.gloasVerificationLeafCursor = f.gloasVerificationLeafCursor.Next() + } + root := f.gloasVerificationLeafCursor.Value.(common.Hash) + if root != (common.Hash{}) { + leaves = append(leaves, root) + } + } + return leaves +} + func (f *ForkChoiceStore) Synced() bool { return f.synced.Load() } @@ -1057,14 +1128,18 @@ func (f *ForkChoiceStore) addPendingELPayload(block *cltypes.SignedBeaconBlock, 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 _, exists := f.pendingELPayloadRoots[root]; exists { + return + } + if f.pendingELPayloadRoots == nil { + f.pendingELPayloadRoots = make(map[common.Hash]struct{}) } } if len(f.pendingELPayloads) >= maxPendingELPayloads { log.Warn("addPendingELPayload: dropping oldest pending EL payload", "queueLen", len(f.pendingELPayloads)) + if evictedRoot, exists := pendingELPayloadRoot(f.pendingELPayloads[0]); exists { + delete(f.pendingELPayloadRoots, evictedRoot) + } copy(f.pendingELPayloads, f.pendingELPayloads[1:]) f.pendingELPayloads[len(f.pendingELPayloads)-1] = PendingELPayload{} f.pendingELPayloads = f.pendingELPayloads[:len(f.pendingELPayloads)-1] @@ -1073,6 +1148,16 @@ func (f *ForkChoiceStore) addPendingELPayload(block *cltypes.SignedBeaconBlock, Block: block, Envelope: envelope, }) + if ok { + f.pendingELPayloadRoots[root] = struct{}{} + } +} + +func (f *ForkChoiceStore) hasPendingELPayload(root common.Hash) bool { + f.pendingELPayloadsMu.Lock() + defer f.pendingELPayloadsMu.Unlock() + _, ok := f.pendingELPayloadRoots[root] + return ok } func pendingELPayloadRoot(p PendingELPayload) (common.Hash, bool) { @@ -1094,16 +1179,19 @@ func (f *ForkChoiceStore) DrainPendingELPayloads() []PendingELPayload { f.pendingELPayloadsMu.Lock() defer f.pendingELPayloadsMu.Unlock() if len(f.pendingELPayloads) == 0 { + clear(f.pendingELPayloadRoots) return nil } if cap(f.pendingELPayloads) > pendingELPayloadsShrinkCap { result := f.pendingELPayloads f.pendingELPayloads = nil + f.pendingELPayloadRoots = nil return result } result := make([]PendingELPayload, len(f.pendingELPayloads)) copy(result, f.pendingELPayloads) clear(f.pendingELPayloads) f.pendingELPayloads = f.pendingELPayloads[:0] + clear(f.pendingELPayloadRoots) return result } diff --git a/cl/phase1/forkchoice/forkchoice_test.go b/cl/phase1/forkchoice/forkchoice_test.go index c11c1784ed4..d2450c5c3fe 100644 --- a/cl/phase1/forkchoice/forkchoice_test.go +++ b/cl/phase1/forkchoice/forkchoice_test.go @@ -37,6 +37,41 @@ import ( "github.com/erigontech/erigon/common" ) +func TestGloasVerificationLeavesPagesAndWraps(t *testing.T) { + root1 := common.HexToHash("0x01") + root3 := common.HexToHash("0x03") + root5 := common.HexToHash("0x05") + root7 := common.HexToHash("0x07") + store := &ForkChoiceStore{headSet: make(map[common.Hash]struct{})} + store.initializeGloasVerificationLeaves() + for _, root := range []common.Hash{{}, root1, root3, root5, root7} { + store.headSet[root] = struct{}{} + store.addGloasVerificationLeaf(root) + } + + require.Nil(t, store.GloasVerificationLeaves(0)) + require.Equal(t, []common.Hash{root1, root3}, store.GloasVerificationLeaves(2)) + require.Equal(t, []common.Hash{root5, root7}, store.GloasVerificationLeaves(2)) + require.Equal(t, []common.Hash{root1, root3}, store.GloasVerificationLeaves(2)) + + store.removeGloasVerificationLeaf(root3) + require.Equal(t, []common.Hash{root5, root7}, store.GloasVerificationLeaves(2)) +} + +func TestGloasVerificationLeavesLargeSetAdvancesFromCursor(t *testing.T) { + store := &ForkChoiceStore{headSet: make(map[common.Hash]struct{})} + store.initializeGloasVerificationLeaves() + for i := uint64(1); i <= 10_000; i++ { + root := testRoot(i) + store.headSet[root] = struct{}{} + store.addGloasVerificationLeaf(root) + } + + require.Equal(t, []common.Hash{testRoot(1), testRoot(2), testRoot(3), testRoot(4), testRoot(5), testRoot(6)}, store.GloasVerificationLeaves(6)) + store.removeGloasVerificationLeaf(testRoot(6)) + require.Equal(t, []common.Hash{testRoot(7), testRoot(8), testRoot(9), testRoot(10), testRoot(11), testRoot(12)}, store.GloasVerificationLeaves(6)) +} + func TestGetFinalizedExecutionHash(t *testing.T) { cache, err := lru.New[common.Hash, common.Hash](16) require.NoError(t, err) diff --git a/cl/phase1/forkchoice/gloas_weight_tree_test.go b/cl/phase1/forkchoice/gloas_weight_tree_test.go index d52bd5b0c17..0c3960b2535 100644 --- a/cl/phase1/forkchoice/gloas_weight_tree_test.go +++ b/cl/phase1/forkchoice/gloas_weight_tree_test.go @@ -544,10 +544,15 @@ func TestOnNewFinalizedPrunesGloasWeightTree(t *testing.T) { defer f.mu.Unlock() f.gloasWeightTree.prepare(justified, cs) require.Contains(t, f.gloasWeightTree.nodes, rootC2) + f.headSet[rootC2] = struct{}{} + f.addGloasVerificationLeaf(rootC2) + require.Contains(t, f.gloasVerificationLeafByRoot, rootC2) f.onNewFinalized(solid.Checkpoint{Epoch: 1, Root: rootC2}) require.NotContains(t, f.gloasWeightTree.nodes, rootC2) + require.NotContains(t, f.headSet, rootC2) + require.NotContains(t, f.gloasVerificationLeafByRoot, rootC2) require.True(t, f.gloasWeightTree.allDirty) } diff --git a/cl/phase1/forkchoice/mock_services/forkchoice_mock.go b/cl/phase1/forkchoice/mock_services/forkchoice_mock.go index 5c283b4a6a8..238df343ab6 100644 --- a/cl/phase1/forkchoice/mock_services/forkchoice_mock.go +++ b/cl/phase1/forkchoice/mock_services/forkchoice_mock.go @@ -18,6 +18,7 @@ package mock_services import ( "context" + "sync/atomic" "testing" "go.uber.org/mock/gomock" @@ -75,8 +76,12 @@ type ForkChoiceStorageMock struct { Headers map[common.Hash]*cltypes.BeaconBlockHeader Blocks map[common.Hash]*cltypes.SignedBeaconBlock Envelopes map[common.Hash]*cltypes.SignedExecutionPayloadEnvelope + ReadEnvelopeErr error + ReadEnvelopeCalls atomic.Int32 + HasEnvelopeOverride *bool VerifiedPayloads map[common.Hash]bool OnExecutionPayloadErr error + OnExecutionPayloadFunc func(context.Context, *cltypes.SignedExecutionPayloadEnvelope, bool, bool) error GetBeaconCommitteeMock func(slot, committeeIndex uint64) ([]uint64, error) Pool pool.OperationsPool @@ -352,6 +357,9 @@ func (f *ForkChoiceStorageMock) OnBlock( } func (f *ForkChoiceStorageMock) OnExecutionPayload(ctx context.Context, signedEnvelope *cltypes.SignedExecutionPayloadEnvelope, checkBlobData, validatePayload bool) error { + if f.OnExecutionPayloadFunc != nil { + return f.OnExecutionPayloadFunc(ctx, signedEnvelope, checkBlobData, validatePayload) + } return f.OnExecutionPayloadErr } @@ -439,6 +447,9 @@ func (f *ForkChoiceStorageMock) GetBlock( } func (f *ForkChoiceStorageMock) HasEnvelope(blockRoot common.Hash) bool { + if f.HasEnvelopeOverride != nil { + return *f.HasEnvelopeOverride + } _, ok := f.Envelopes[blockRoot] return ok } @@ -451,7 +462,8 @@ func (f *ForkChoiceStorageMock) IsPayloadVerified(blockRoot common.Hash) bool { } func (f *ForkChoiceStorageMock) ReadEnvelopeFromDisk(blockRoot common.Hash) (*cltypes.SignedExecutionPayloadEnvelope, error) { - return f.Envelopes[blockRoot], nil + f.ReadEnvelopeCalls.Add(1) + return f.Envelopes[blockRoot], f.ReadEnvelopeErr } func (f *ForkChoiceStorageMock) IsBlobDataAvailable(slot uint64, blockRoot common.Hash) bool { diff --git a/cl/phase1/forkchoice/on_block.go b/cl/phase1/forkchoice/on_block.go index a31e3b4cff1..d0fa4adc3b4 100644 --- a/cl/phase1/forkchoice/on_block.go +++ b/cl/phase1/forkchoice/on_block.go @@ -49,6 +49,7 @@ const foreseenProposers = 16 var ( ErrEIP4844DataNotAvailable = errors.New("EIP-4844 blob data is not available") ErrEIP7594ColumnDataNotAvailable = errors.New("EIP-7594 column data is not available") + ErrExecutionPayloadAlreadyStored = errors.New("execution payload envelope already stored") ErrNewPayloadNoStatus = errors.New("newPayload returned no status") ErrMissingSegment = errors.New("missing segment: parent state not available") ErrParentEnvelopePending = errors.New("parent execution payload envelope not yet available") @@ -324,7 +325,9 @@ func (f *ForkChoiceStore) OnBlock(ctx context.Context, block *cltypes.SignedBeac // Remove the parent from the head set delete(f.headSet, block.Block.ParentRoot) + f.removeGloasVerificationLeaf(block.Block.ParentRoot) f.headSet[blockRoot] = struct{}{} + f.addGloasVerificationLeaf(blockRoot) // record_block_timeliness: store [block_timely, ptc_timely] vector. // [Modified in Gloas:EIP7732] Post-GLOAS stores a two-element timeliness vector; // pre-GLOAS stores [block_timely, false]. See recordBlockTimeliness for details. diff --git a/cl/phase1/forkchoice/on_execution_payload.go b/cl/phase1/forkchoice/on_execution_payload.go index 780855a1cbd..0c8297200a0 100644 --- a/cl/phase1/forkchoice/on_execution_payload.go +++ b/cl/phase1/forkchoice/on_execution_payload.go @@ -297,6 +297,7 @@ func (f *ForkChoiceStore) validatePayloadWithEL( if err := f.optimisticStore.AddOptimisticCandidate(beaconBlockRoot, block.Block); err != nil { return fmt.Errorf("failed to add block to optimistic store: %v", err) } + return errELBehind case execution_client.PayloadStatusInvalidated: log.Warn("validatePayloadWithEL: payload is invalid", "beaconBlockRoot", beaconBlockRoot, "err", err) f.markPayloadInvalidLocked(beaconBlockRoot, executionBlockHash) @@ -495,9 +496,12 @@ func (f *ForkChoiceStore) OnExecutionPayload(ctx context.Context, signedEnvelope // Process envelope under f.mu; DB index write happens after unlock to avoid // deadlock with postForkchoiceOperations (which holds MDBX tx then needs f.mu.RLock). applied, err := f.applyEnvelope(ctx, signedEnvelope, checkBlobData, validatePayload) - if err != nil || !applied { + if err != nil { return err } + if !applied { + return fmt.Errorf("%w: %w", ErrIgnore, ErrExecutionPayloadAlreadyStored) + } // Write execution block indices outside f.mu. if f.db != nil { diff --git a/cl/phase1/forkchoice/on_execution_payload_test.go b/cl/phase1/forkchoice/on_execution_payload_test.go index 85dd3524a94..173fb04b9ff 100644 --- a/cl/phase1/forkchoice/on_execution_payload_test.go +++ b/cl/phase1/forkchoice/on_execution_payload_test.go @@ -29,6 +29,7 @@ import ( "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/cl/phase1/forkchoice/optimistic" "github.com/erigontech/erigon/common" ) @@ -62,6 +63,17 @@ func TestValidateEnvelopeAgainstBlock_NoBid(t *testing.T) { require.Contains(t, err.Error(), "block missing signed_execution_payload_bid") } +func TestOnExecutionPayloadReportsAlreadyStored(t *testing.T) { + f := &ForkChoiceStore{forkGraph: payloadVoteForkGraph{hasEnvelope: true}} + envelope := &cltypes.SignedExecutionPayloadEnvelope{ + Message: &cltypes.ExecutionPayloadEnvelope{BeaconBlockRoot: common.HexToHash("0x1234")}, + } + + err := f.OnExecutionPayload(t.Context(), envelope, true, true) + require.ErrorIs(t, err, ErrExecutionPayloadAlreadyStored) + require.ErrorIs(t, err, ErrIgnore) +} + // TestValidateEnvelopeAgainstBlock_SlotNumberMismatch tests that validation fails when // block.slot != envelope.payload.slot_number (EIP-7843 / GLOAS p2p-interface REJECT rule). func TestValidateEnvelopeAgainstBlock_SlotNumberMismatch(t *testing.T) { @@ -289,6 +301,11 @@ func TestValidatePayloadWithELDoesNotRelockForkChoiceMu(t *testing.T) { status: execution_client.PayloadStatusValidated, wantVerify: true, }, + { + name: "not validated", + status: execution_client.PayloadStatusNotValidated, + wantErr: true, + }, { name: "invalidated", status: execution_client.PayloadStatusInvalidated, @@ -322,6 +339,7 @@ func TestValidatePayloadWithELDoesNotRelockForkChoiceMu(t *testing.T) { executionPayloadStatus: executionPayloadStatus, payloadStatusByRoot: payloadStatusByRoot, executionPayloadGasLimit: executionPayloadGasLimit, + optimisticStore: optimistic.NewOptimisticStore(), } envelope := &cltypes.ExecutionPayloadEnvelope{ Payload: &cltypes.Eth1Block{BlockHash: executionBlockHash}, diff --git a/cl/phase1/forkchoice/payload_vote.go b/cl/phase1/forkchoice/payload_vote.go index dee2b4df76a..b2e0342d45d 100644 --- a/cl/phase1/forkchoice/payload_vote.go +++ b/cl/phase1/forkchoice/payload_vote.go @@ -425,14 +425,20 @@ func (f *ForkChoiceStore) validateParentPayloadPath(block *cltypes.BeaconBlock) } if f.isParentNodeFull(block) { - // Parent is FULL - verify execution payload envelope exists on disk. - // Return ErrParentEnvelopePending (not a hard error) when the envelope is - // missing. During forward sync the envelope may not yet be persisted (it - // arrives in the same batch or in a later batch), so a hard error would - // permanently reject the block and ban the peer. + // A FULL parent remains pending until its payload is persisted and verified by any configured EL. if !f.forkGraph.HasEnvelope(block.ParentRoot) { return ErrParentEnvelopePending } + if f.engine != nil && !f.IsPayloadVerified(block.ParentRoot) { + if !f.hasPendingELPayload(block.ParentRoot) { + parentBlock, ok := f.forkGraph.GetBlock(block.ParentRoot) + envelope, err := f.forkGraph.ReadEnvelopeFromDisk(block.ParentRoot) + if ok && parentBlock != nil && err == nil && envelope != nil && envelope.Message != nil { + f.addPendingELPayload(parentBlock, envelope) + } + } + return ErrParentEnvelopePending + } } else { // Parent is EMPTY - verify bid.parent_block_hash == parent_bid.parent_block_hash parentBlock, ok := f.forkGraph.GetBlock(block.ParentRoot) diff --git a/cl/phase1/forkchoice/payload_vote_test.go b/cl/phase1/forkchoice/payload_vote_test.go index 5443359aadd..1155dfd57b4 100644 --- a/cl/phase1/forkchoice/payload_vote_test.go +++ b/cl/phase1/forkchoice/payload_vote_test.go @@ -5,6 +5,7 @@ import ( lru "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" @@ -34,6 +35,9 @@ func (g ptcVoteForkGraph) GetBlock(root common.Hash) (*cltypes.SignedBeaconBlock type payloadVoteForkGraph struct { fork_graph.ForkGraph hasEnvelope bool + blocks map[common.Hash]*cltypes.SignedBeaconBlock + envelope *cltypes.SignedExecutionPayloadEnvelope + readEnvelopeCalls *int dumpedEnvelope *common.Hash invalidatedHeader *common.Hash } @@ -42,6 +46,18 @@ func (g payloadVoteForkGraph) HasEnvelope(common.Hash) bool { return g.hasEnvelope } +func (g payloadVoteForkGraph) GetBlock(root common.Hash) (*cltypes.SignedBeaconBlock, bool) { + block, ok := g.blocks[root] + return block, ok +} + +func (g payloadVoteForkGraph) ReadEnvelopeFromDisk(common.Hash) (*cltypes.SignedExecutionPayloadEnvelope, error) { + if g.readEnvelopeCalls != nil { + (*g.readEnvelopeCalls)++ + } + return g.envelope, nil +} + func (g payloadVoteForkGraph) DumpEnvelopeOnDisk(blockRoot common.Hash, _ *cltypes.SignedExecutionPayloadEnvelope) error { if g.dumpedEnvelope != nil { *g.dumpedEnvelope = blockRoot @@ -340,6 +356,62 @@ func TestGloasForkChoiceRequiresVerifiedPayload(t *testing.T) { } } +func TestValidateParentPayloadPathRequiresVerifiedFullParent(t *testing.T) { + parentRoot := common.HexToHash("0x1234") + parentBlockHash := common.HexToHash("0xabcd") + parent := cltypes.NewSignedBeaconBlock(&clparams.MainnetBeaconConfig, clparams.GloasVersion) + parent.Block.Body.SignedExecutionPayloadBid.Message.BlockHash = parentBlockHash + child := cltypes.NewBeaconBlock(&clparams.MainnetBeaconConfig, clparams.GloasVersion) + child.ParentRoot = parentRoot + child.Body.SignedExecutionPayloadBid.Message.ParentBlockHash = parentBlockHash + + for _, tt := range []struct { + name string + withEL bool + verified bool + wantErr bool + }{ + {name: "persisted but unverified", withEL: true, wantErr: true}, + {name: "persisted and verified", withEL: true, verified: true}, + {name: "standalone without EL"}, + {name: "standalone with prior verification", verified: true}, + } { + t.Run(tt.name, func(t *testing.T) { + readEnvelopeCalls := 0 + f := newPayloadVoteTestStore(t, parentRoot, true, tt.verified) + if tt.withEL { + f.engine = execution_client.NewMockExecutionEngine(gomock.NewController(t)) + } + f.forkGraph = payloadVoteForkGraph{ + hasEnvelope: true, + blocks: map[common.Hash]*cltypes.SignedBeaconBlock{parentRoot: parent}, + readEnvelopeCalls: &readEnvelopeCalls, + envelope: &cltypes.SignedExecutionPayloadEnvelope{ + Message: &cltypes.ExecutionPayloadEnvelope{ + BeaconBlockRoot: parentRoot, + Payload: &cltypes.Eth1Block{}, + }, + }, + } + + err := f.validateParentPayloadPath(child) + if tt.wantErr { + require.ErrorIs(t, err, ErrParentEnvelopePending) + require.ErrorIs(t, f.validateParentPayloadPath(child), ErrParentEnvelopePending) + if tt.withEL { + require.Equal(t, 1, readEnvelopeCalls) + require.Len(t, f.DrainPendingELPayloads(), 1) + } else { + require.Zero(t, readEnvelopeCalls) + require.Empty(t, f.DrainPendingELPayloads()) + } + } else { + require.NoError(t, err) + } + }) + } +} + func TestIsPayloadVerifiedStrictSemantics(t *testing.T) { root := common.HexToHash("0x5678") diff --git a/cl/phase1/forkchoice/pending_el_payload_test.go b/cl/phase1/forkchoice/pending_el_payload_test.go index e4c5c9dd996..a7222499401 100644 --- a/cl/phase1/forkchoice/pending_el_payload_test.go +++ b/cl/phase1/forkchoice/pending_el_payload_test.go @@ -50,4 +50,22 @@ func TestPendingELPayloadsDeduplicateByEnvelopeRoot(t *testing.T) { payloads := f.DrainPendingELPayloads() require.Len(t, payloads, 1) require.Equal(t, uint64(1), payloads[0].Block.Block.Slot) + require.False(t, f.hasPendingELPayload(root)) +} + +func TestPendingELPayloadsEvictionUpdatesRootMembership(t *testing.T) { + f := &ForkChoiceStore{} + firstRoot := common.Hash{31: 1} + + for i := range maxPendingELPayloads + 1 { + root := common.Hash{30: byte((i + 1) >> 8), 31: byte(i + 1)} + f.addPendingELPayload(nil, &cltypes.SignedExecutionPayloadEnvelope{ + Message: &cltypes.ExecutionPayloadEnvelope{BeaconBlockRoot: root}, + }) + } + + require.False(t, f.hasPendingELPayload(firstRoot)) + require.True(t, f.hasPendingELPayload(common.Hash{ + 30: byte((maxPendingELPayloads + 1) >> 8), 31: byte((maxPendingELPayloads + 1) & 0xff), + })) } diff --git a/cl/phase1/forkchoice/utils.go b/cl/phase1/forkchoice/utils.go index a6625dad4fc..f907435d3cc 100644 --- a/cl/phase1/forkchoice/utils.go +++ b/cl/phase1/forkchoice/utils.go @@ -154,10 +154,24 @@ func (f *ForkChoiceStore) onNewFinalized(newFinalized solid.Checkpoint) { f.childrens.Range(func(k, v any) bool { if v.(childrens).parentSlot <= finalizedSlot { f.childrens.Delete(k) - delete(f.headSet, k.(common.Hash)) + root := k.(common.Hash) + delete(f.headSet, root) + f.removeGloasVerificationLeaf(root) } return true }) + if f.gloasVerificationLeaves != nil { + for element := f.gloasVerificationLeaves.Front(); element != nil; { + next := element.Next() + root := element.Value.(common.Hash) + header, ok := f.forkGraph.GetHeader(root) + if !ok || header.Slot <= finalizedSlot { + delete(f.headSet, root) + f.removeGloasVerificationLeaf(root) + } + element = next + } + } // Clean up per-block unrealized justifications/finalizations for finalized blocks. f.unrealizedJustifications.Range(func(k, v any) bool { diff --git a/cl/phase1/network/services/execution_payload_service.go b/cl/phase1/network/services/execution_payload_service.go index cdac0bf5c17..beec8692d30 100644 --- a/cl/phase1/network/services/execution_payload_service.go +++ b/cl/phase1/network/services/execution_payload_service.go @@ -41,28 +41,32 @@ type seenEnvelopeKey struct { builderIndex uint64 } -// pendingEnvelopeKey tracks envelopes waiting for their block to arrive. -// We use (blockRoot, envelopeHash) as key instead of just blockRoot because: -// - Multiple envelopes (including forged ones) may arrive before the block -// - Using only blockRoot would cause later arrivals to overwrite earlier ones -// - If a forged envelope overwrites the valid one, we lose the valid envelope -// - With envelopeHash, all candidates are kept and validated when block arrives +// pendingEnvelopeKey retains distinct unvalidated candidates; DA-validated jobs use one entry per block root. type pendingEnvelopeKey struct { - blockRoot common.Hash - envelopeHash common.Hash + blockRoot common.Hash + envelopeHash common.Hash + dataAvailability bool } -// envelopeJob represents a pending envelope waiting for its block to arrive +// envelopeJob represents an envelope waiting for its block or PeerDAS data. type envelopeJob struct { envelope *cltypes.SignedExecutionPayloadEnvelope creationTime time.Time + processing bool + recovered atomic.Bool + validate atomic.Bool + nextAttempt time.Time + retryDelay time.Duration } const ( - seenEnvelopeCacheSize = 1000 - pendingEnvelopeExpiry = 30 * time.Second - pendingEnvelopeCheckInterval = 100 * time.Millisecond - maxPendingEnvelopes = 1024 + seenEnvelopeCacheSize = 1000 + pendingEnvelopeExpiry = 30 * time.Second + pendingDataAvailabilityExpiry = 2 * time.Minute + pendingEnvelopeCheckInterval = 100 * time.Millisecond + pendingEnvelopeInitialRetry = time.Second + pendingEnvelopeMaxRetry = 10 * time.Second + maxPendingEnvelopes = 1024 ) type executionPayloadService struct { @@ -73,8 +77,9 @@ type executionPayloadService struct { // Cache to track seen envelopes: (beaconBlockRoot, builderIndex) -> struct{} seenEnvelopesCache *lru.Cache[seenEnvelopeKey, struct{}] - // Pending envelopes waiting for block to arrive + // pendingMu keeps map membership and count changes atomic; retry timing is owned by loop. pendingEnvelopes sync.Map // pendingEnvelopeKey -> *envelopeJob + pendingMu sync.Mutex pendingCount atomic.Int32 pendingCond *sync.Cond } @@ -123,6 +128,14 @@ func (s *executionPayloadService) DecodeGossipMessage(_ peer.ID, data []byte, ve // Reference: https://github.com/ethereum/consensus-specs/blob/dev/specs/_features/epbs/p2p-interface.md#execution_payload // [New in Gloas:EIP7732] func (s *executionPayloadService) ProcessMessage(ctx context.Context, _ *uint64, signedEnvelope *cltypes.SignedExecutionPayloadEnvelope) error { + return s.processEnvelope(ctx, signedEnvelope, false, true, true) +} + +func (s *executionPayloadService) ProcessRecoveredEnvelope(ctx context.Context, signedEnvelope *cltypes.SignedExecutionPayloadEnvelope, validatePayload bool) error { + return s.processEnvelope(ctx, signedEnvelope, true, validatePayload, true) +} + +func (s *executionPayloadService) processEnvelope(ctx context.Context, signedEnvelope *cltypes.SignedExecutionPayloadEnvelope, recovered, validatePayload, queueOnRetry bool) error { if signedEnvelope == nil || signedEnvelope.Message == nil { return errors.New("nil execution payload envelope") } @@ -139,21 +152,20 @@ func (s *executionPayloadService) ProcessMessage(ctx context.Context, _ *uint64, // A client MAY queue payload for processing once the block is retrieved. block, ok := s.forkchoiceStore.GetBlock(beaconBlockRoot) if !ok || block == nil { - // Block hasn't arrived yet, queue envelope for later processing - s.queuePendingEnvelope(beaconBlockRoot, signedEnvelope) - // Also store in forkchoice's pendingEnvelopes so OnBlock can process it immediately - // when the block arrives, instead of waiting for the 100ms polling loop. - // validatePayload must be true: if the block arrives (via OnBlock) before this call - // acquires f.mu, the envelope will be applied with validatePayload — ensuring - // NewPayload is sent to the EL. With false, a mutex-contention race silently - // marks the envelope as processed without ever notifying the EL, permanently - // breaking the chain. - s.forkchoiceStore.OnExecutionPayload(ctx, signedEnvelope, false, true) + if queueOnRetry { + s.queuePendingEnvelopeWithOptions(beaconBlockRoot, signedEnvelope, recovered, validatePayload, false) + } + if !recovered { + s.forkchoiceStore.OnExecutionPayload(ctx, signedEnvelope, false, true) + } log.Trace("Queued execution payload envelope for later processing", "beaconBlockRoot", beaconBlockRoot, "builderIndex", builderIndex) return ErrIgnore } + if block.Block == nil { + return errors.New("nil beacon block") + } // [IGNORE] The node has not seen another valid SignedExecutionPayloadEnvelope // for this block root from this builder. @@ -161,36 +173,39 @@ func (s *executionPayloadService) ProcessMessage(ctx context.Context, _ *uint64, beaconBlockRoot: beaconBlockRoot, builderIndex: builderIndex, } - if s.seenEnvelopesCache.Contains(seenKey) { + if !recovered && s.seenEnvelopesCache.Contains(seenKey) { return fmt.Errorf("%w: already seen envelope for block %v from builder %d", ErrIgnore, beaconBlockRoot, builderIndex) } // [IGNORE] The envelope is from a slot greater than or equal to the latest finalized slot - finalizedSlot := s.forkchoiceStore.FinalizedSlot() - if block.Block.Slot < finalizedSlot { - return fmt.Errorf("%w: envelope slot %d < finalized slot %d", ErrIgnore, block.Block.Slot, finalizedSlot) + if !recovered { + finalizedSlot := s.forkchoiceStore.FinalizedSlot() + if block.Block.Slot < finalizedSlot { + return fmt.Errorf("%w: envelope slot %d < finalized slot %d", ErrIgnore, block.Block.Slot, finalizedSlot) + } + } + if !recovered && s.forkchoiceStore.HasEnvelope(beaconBlockRoot) { + return storedEnvelopeResult(beaconBlockRoot, block, builderIndex) } // Process the execution payload through forkchoice // Note: bid matching and signature verification are done in OnExecutionPayload.validateEnvelopeAgainstBlock - if err := s.forkchoiceStore.OnExecutionPayload(ctx, signedEnvelope, true, true); err != nil { + if err := s.forkchoiceStore.OnExecutionPayload(ctx, signedEnvelope, true, validatePayload); err != nil { + if errors.Is(err, forkchoice.ErrExecutionPayloadAlreadyStored) { + return storedEnvelopeResult(beaconBlockRoot, block, builderIndex) + } if errors.Is(err, forkchoice.ErrIgnore) || errors.Is(err, forkchoice.ErrEIP7594ColumnDataNotAvailable) { - return fmt.Errorf("%w: %v", ErrIgnore, err) + if queueOnRetry { + s.queuePendingEnvelopeWithOptions(beaconBlockRoot, signedEnvelope, recovered, validatePayload, errors.Is(err, forkchoice.ErrEIP7594ColumnDataNotAvailable)) + } + return fmt.Errorf("%w: %w", ErrIgnore, err) } return fmt.Errorf("failed to process execution payload: %w", err) } - // Mark as seen AFTER successful validation - // This ensures invalid envelopes (e.g., with forged signatures) don't block valid ones - s.seenEnvelopesCache.Add(seenKey, struct{}{}) + s.markEnvelopeAvailable(seenKey, block.Block.Slot) - // Emit SSE event for execution_payload_available [New in Gloas:EIP7732] - s.emitters.Operation().SendExecutionPayloadAvailable(&beaconevents.ExecutionPayloadAvailableData{ - Slot: block.Block.Slot, - BlockRoot: beaconBlockRoot, - }) - - log.Trace("Processed execution payload via gossip", + log.Trace("Processed execution payload envelope", "slot", block.Block.Slot, "beaconBlockRoot", beaconBlockRoot, "builderIndex", builderIndex) @@ -198,36 +213,190 @@ func (s *executionPayloadService) ProcessMessage(ctx context.Context, _ *uint64, return nil } +func (s *executionPayloadService) markEnvelopeAvailable(key seenEnvelopeKey, slot uint64) { + if seen, _ := s.seenEnvelopesCache.ContainsOrAdd(key, struct{}{}); seen { + return + } + s.emitters.Operation().SendExecutionPayloadAvailable(&beaconevents.ExecutionPayloadAvailableData{ + Slot: slot, + BlockRoot: key.beaconBlockRoot, + }) +} + +func (s *executionPayloadService) accountStoredPendingEnvelope(block *cltypes.SignedBeaconBlock, key pendingEnvelopeKey, job *envelopeJob) bool { + stored, err := s.forkchoiceStore.ReadEnvelopeFromDisk(key.blockRoot) + if err != nil || stored == nil || stored.Message == nil || job.envelope == nil || job.envelope.Message == nil { + return false + } + if stored.Message.BuilderIndex != job.envelope.Message.BuilderIndex { + return false + } + s.markEnvelopeAvailable(seenEnvelopeKey{ + beaconBlockRoot: key.blockRoot, + builderIndex: stored.Message.BuilderIndex, + }, block.Block.Slot) + s.removePendingEnvelope(key, job) + return true +} + +func storedEnvelopeResult(blockRoot common.Hash, block *cltypes.SignedBeaconBlock, builderIndex uint64) error { + if block == nil || block.Block == nil || block.Block.Body == nil { + return fmt.Errorf("%w: stored envelope block is incomplete", ErrIgnore) + } + bid := block.Block.Body.GetSignedExecutionPayloadBid() + if bid == nil || bid.Message == nil { + return fmt.Errorf("%w: stored envelope block has no committed bid", ErrIgnore) + } + storedBuilder := bid.Message.BuilderIndex + if storedBuilder != builderIndex { + return fmt.Errorf("envelope builder_index %d != stored builder_index %d", builderIndex, storedBuilder) + } + return fmt.Errorf("%w: envelope already applied for block %v from builder %d", ErrIgnore, blockRoot, builderIndex) +} + // queuePendingEnvelope adds an envelope to the pending queue for later processing func (s *executionPayloadService) queuePendingEnvelope(blockRoot common.Hash, envelope *cltypes.SignedExecutionPayloadEnvelope) { - if s.pendingCount.Add(1) > maxPendingEnvelopes { - s.pendingCount.Add(-1) + s.queuePendingEnvelopeWithOptions(blockRoot, envelope, false, true, false) +} + +func (s *executionPayloadService) queuePendingEnvelopeWithOptions(blockRoot common.Hash, envelope *cltypes.SignedExecutionPayloadEnvelope, recovered, validatePayload, dataAvailability bool) { + key := pendingEnvelopeKey{blockRoot: blockRoot, dataAvailability: dataAvailability} + + var envelopeHash common.Hash + if !dataAvailability { + if !recovered && s.pendingCount.Load() >= maxPendingEnvelopes { + return + } + var err error + envelopeHash, err = envelope.HashSSZ() + if err != nil { + log.Warn("Failed to hash envelope for pending queue", "blockRoot", blockRoot, "err", err) + return + } + } + key.envelopeHash = envelopeHash + + job := &envelopeJob{ + envelope: envelope, + creationTime: time.Now(), + } + job.recovered.Store(recovered) + job.validate.Store(validatePayload) + if dataAvailability { + job.retryDelay = pendingEnvelopeInitialRetry + job.nextAttempt = time.Now().Add(job.retryDelay) + } + + s.pendingMu.Lock() + if actual, loaded := s.pendingEnvelopes.Load(key); loaded { + storedJob := actual.(*envelopeJob) + upgradeEnvelopeJob(storedJob, recovered, validatePayload) + if dataAvailability { + storedJob.creationTime = time.Now() + } + s.pendingMu.Unlock() return } + if dataAvailability && s.pendingCount.Load() >= maxPendingEnvelopes { + s.evictUnvalidatedPendingEnvelope() + } + if s.pendingCount.Load() >= maxPendingEnvelopes { + s.pendingMu.Unlock() + return + } + s.pendingEnvelopes.Store(key, job) + s.pendingCount.Add(1) + s.pendingMu.Unlock() - // Compute envelope hash to allow multiple candidates per block - envelopeHash, err := envelope.HashSSZ() - if err != nil { - s.pendingCount.Add(-1) - log.Warn("Failed to hash envelope for pending queue", "blockRoot", blockRoot, "err", err) + s.pendingCond.L.Lock() + s.pendingCond.Signal() + s.pendingCond.L.Unlock() +} + +func (s *executionPayloadService) evictUnvalidatedPendingEnvelope() { + s.pendingEnvelopes.Range(func(key, value any) bool { + pendingKey := key.(pendingEnvelopeKey) + job := value.(*envelopeJob) + if pendingKey.dataAvailability || job.processing { + return true + } + if s.pendingEnvelopes.CompareAndDelete(pendingKey, value) { + s.pendingCount.Add(-1) + } + return false + }) +} + +func (s *executionPayloadService) claimPendingEnvelope(key pendingEnvelopeKey, job *envelopeJob) bool { + s.pendingMu.Lock() + defer s.pendingMu.Unlock() + actual, ok := s.pendingEnvelopes.Load(key) + if !ok || actual != job || job.processing { + return false + } + job.processing = true + return true +} + +func (s *executionPayloadService) releasePendingEnvelope(job *envelopeJob) { + s.pendingMu.Lock() + job.processing = false + s.pendingMu.Unlock() +} + +func upgradeEnvelopeJob(job *envelopeJob, recovered, validatePayload bool) { + if validatePayload { + job.validate.Store(true) + } + if recovered { + job.recovered.Store(true) + } +} + +func (s *executionPayloadService) promoteDataAvailabilityRetry(oldKey pendingEnvelopeKey, job *envelopeJob) { + job.retryDelay = pendingEnvelopeInitialRetry + job.nextAttempt = time.Now().Add(job.retryDelay) + newKey := pendingEnvelopeKey{blockRoot: oldKey.blockRoot, dataAvailability: true} + s.pendingMu.Lock() + if actual, ok := s.pendingEnvelopes.Load(oldKey); !ok || actual != job { + s.pendingMu.Unlock() return } + actual, loaded := s.pendingEnvelopes.Load(newKey) + if loaded { + storedJob := actual.(*envelopeJob) + upgradeEnvelopeJob(storedJob, job.recovered.Load(), job.validate.Load()) + storedJob.creationTime = time.Now() + } else { + job.creationTime = time.Now() + s.pendingEnvelopes.Store(newKey, job) + } + if s.pendingEnvelopes.CompareAndDelete(oldKey, job) && loaded { + s.pendingCount.Add(-1) + } + s.pendingMu.Unlock() +} - key := pendingEnvelopeKey{ - blockRoot: blockRoot, - envelopeHash: envelopeHash, +func (s *executionPayloadService) removePendingEnvelope(key pendingEnvelopeKey, job *envelopeJob) { + s.pendingMu.Lock() + if s.pendingEnvelopes.CompareAndDelete(key, job) { + s.pendingCount.Add(-1) } + s.pendingMu.Unlock() +} - if _, loaded := s.pendingEnvelopes.LoadOrStore(key, &envelopeJob{ - envelope: envelope, - creationTime: time.Now(), - }); loaded { +func (s *executionPayloadService) expirePendingEnvelope(key pendingEnvelopeKey, job *envelopeJob, expiry time.Duration) bool { + s.pendingMu.Lock() + defer s.pendingMu.Unlock() + actual, ok := s.pendingEnvelopes.Load(key) + if !ok || actual != job || time.Since(job.creationTime) <= expiry { + return false + } + if s.pendingEnvelopes.CompareAndDelete(key, job) { s.pendingCount.Add(-1) - } else { - s.pendingCond.L.Lock() - s.pendingCond.Signal() - s.pendingCond.L.Unlock() + return true } + return false } // loop is the background goroutine that processes pending envelopes @@ -277,9 +446,11 @@ func (s *executionPayloadService) processPendingEnvelopes(ctx context.Context) { job := value.(*envelopeJob) // Check expiry - if time.Since(job.creationTime) > pendingEnvelopeExpiry { - s.pendingEnvelopes.Delete(pendingKey) - s.pendingCount.Add(-1) + expiry := pendingEnvelopeExpiry + if pendingKey.dataAvailability { + expiry = pendingDataAvailabilityExpiry + } + if s.expirePendingEnvelope(pendingKey, job, expiry) { log.Trace("Pending envelope expired", "blockRoot", pendingKey.blockRoot) return true } @@ -289,13 +460,32 @@ func (s *executionPayloadService) processPendingEnvelopes(ctx context.Context) { if !ok || block == nil { return true // Block still not here, keep waiting } + if pendingKey.dataAvailability && time.Now().Before(job.nextAttempt) { + return true + } + if !s.claimPendingEnvelope(pendingKey, job) { + return true + } + defer s.releasePendingEnvelope(job) - // Block arrived, remove from pending and process - s.pendingEnvelopes.Delete(pendingKey) - s.pendingCount.Add(-1) - - // Re-run full validation via ProcessMessage - if err := s.ProcessMessage(ctx, nil, job.envelope); err != nil { + err := s.processEnvelope(ctx, job.envelope, job.recovered.Load(), job.validate.Load(), false) + if errors.Is(err, forkchoice.ErrEIP7594ColumnDataNotAvailable) { + if !pendingKey.dataAvailability { + s.promoteDataAvailabilityRetry(pendingKey, job) + return true + } + job.retryDelay = min(job.retryDelay*2, pendingEnvelopeMaxRetry) + job.nextAttempt = time.Now().Add(job.retryDelay) + return true + } + if errors.Is(err, ErrIgnore) && s.accountStoredPendingEnvelope(block, pendingKey, job) { + return true + } + if errors.Is(err, forkchoice.ErrIgnore) || errors.Is(err, ErrIgnore) { + return true + } + s.removePendingEnvelope(pendingKey, job) + if err != nil { log.Trace("Failed to process pending envelope", "blockRoot", pendingKey.blockRoot, "err", err) } return true diff --git a/cl/phase1/network/services/execution_payload_service_test.go b/cl/phase1/network/services/execution_payload_service_test.go index 9e32176d091..7fa0461d448 100644 --- a/cl/phase1/network/services/execution_payload_service_test.go +++ b/cl/phase1/network/services/execution_payload_service_test.go @@ -30,6 +30,7 @@ import ( "github.com/erigontech/erigon/cl/cltypes" "github.com/erigontech/erigon/cl/cltypes/solid" "github.com/erigontech/erigon/cl/phase1/core/state/lru" + "github.com/erigontech/erigon/cl/phase1/forkchoice" "github.com/erigontech/erigon/cl/phase1/forkchoice/mock_services" "github.com/erigontech/erigon/common" ) @@ -41,6 +42,20 @@ func setupExecutionPayloadService(t *testing.T) (ExecutionPayloadService, *mock_ return service, forkchoiceMock } +func setupExecutionPayloadServiceWithoutLoop(t *testing.T) (*executionPayloadService, *mock_services.ForkChoiceStorageMock) { + cfg := &clparams.MainnetBeaconConfig + forkchoiceMock := mock_services.NewForkChoiceStorageMock(t) + seenCache, err := lru.New[seenEnvelopeKey, struct{}]("seen_envelopes", seenEnvelopeCacheSize) + require.NoError(t, err) + return &executionPayloadService{ + forkchoiceStore: forkchoiceMock, + beaconCfg: cfg, + emitters: beaconevents.NewEventEmitter(), + seenEnvelopesCache: seenCache, + pendingCond: sync.NewCond(&sync.Mutex{}), + }, forkchoiceMock +} + func newTestSignedEnvelope(slot uint64, blockRoot common.Hash, builderIndex uint64) *cltypes.SignedExecutionPayloadEnvelope { envelope := cltypes.NewExecutionPayloadEnvelope(&clparams.MainnetBeaconConfig) envelope.BeaconBlockRoot = blockRoot @@ -56,6 +71,13 @@ func newTestSignedEnvelope(slot uint64, blockRoot common.Hash, builderIndex uint } } +func newTestSignedBlockWithBuilder(_ common.Hash, slot, builderIndex uint64) *cltypes.SignedBeaconBlock { + block := cltypes.NewSignedBeaconBlock(&clparams.MainnetBeaconConfig, clparams.GloasVersion) + block.Block.Slot = slot + block.Block.Body.SignedExecutionPayloadBid.Message.BuilderIndex = builderIndex + return block +} + func TestExecutionPayloadServiceNilEnvelope(t *testing.T) { service, _ := setupExecutionPayloadService(t) @@ -97,6 +119,35 @@ func TestExecutionPayloadServiceBlockNotFound(t *testing.T) { require.NoError(t, err) } +func TestExecutionPayloadServiceAccountsEnvelopeAppliedWhenBlockArrives(t *testing.T) { + impl, fcu := setupExecutionPayloadServiceWithoutLoop(t) + blockRoot := common.HexToHash("0x1234") + envelope := newTestSignedEnvelope(100, blockRoot, 1) + events := make(chan *beaconevents.EventStream, 1) + sub := impl.emitters.Operation().Subscribe(events) + defer sub.Unsubscribe() + + require.ErrorIs(t, impl.ProcessMessage(t.Context(), nil, envelope), ErrIgnore) + require.Equal(t, int32(1), impl.pendingCount.Load()) + fcu.Blocks[blockRoot] = &cltypes.SignedBeaconBlock{Block: &cltypes.BeaconBlock{Slot: 100}} + fcu.Envelopes[blockRoot] = envelope + impl.processPendingEnvelopes(t.Context()) + + require.Equal(t, int32(0), impl.pendingCount.Load()) + require.True(t, impl.seenEnvelopesCache.Contains(seenEnvelopeKey{blockRoot, 1})) + require.Equal(t, beaconevents.OpExecutionPayloadAvailable, (<-events).Event) +} + +func TestExecutionPayloadServiceNilBeaconBlock(t *testing.T) { + impl, fcu := setupExecutionPayloadServiceWithoutLoop(t) + blockRoot := common.HexToHash("0x1234") + fcu.Blocks[blockRoot] = new(cltypes.SignedBeaconBlock) + + err := impl.ProcessMessage(t.Context(), nil, newTestSignedEnvelope(100, blockRoot, 1)) + require.Error(t, err) + require.Contains(t, err.Error(), "nil beacon block") +} + func TestExecutionPayloadServiceAlreadySeen(t *testing.T) { service, fcu := setupExecutionPayloadService(t) @@ -264,10 +315,12 @@ func TestExecutionPayloadServicePendingEnvelopeProcessing(t *testing.T) { blockRoot: blockRoot, envelopeHash: envelopeHash, } - impl.pendingEnvelopes.Store(key, &envelopeJob{ + job := &envelopeJob{ envelope: envelope, creationTime: time.Now(), - }) + } + job.validate.Store(true) + impl.pendingEnvelopes.Store(key, job) impl.pendingCount.Store(1) // Block not yet available - should keep pending @@ -291,6 +344,208 @@ func TestExecutionPayloadServicePendingEnvelopeProcessing(t *testing.T) { require.True(t, impl.seenEnvelopesCache.Contains(seenEnvelopeKey{blockRoot, 1})) } +func TestExecutionPayloadServiceRetriesEnvelopeUntilColumnDataAvailable(t *testing.T) { + impl, fcu := setupExecutionPayloadServiceWithoutLoop(t) + blockRoot := common.HexToHash("0x1234") + envelope := newTestSignedEnvelope(100, blockRoot, 1) + fcu.Blocks[blockRoot] = newTestSignedBlockWithBuilder(blockRoot, 100, 1) + fcu.OnExecutionPayloadErr = forkchoice.ErrEIP7594ColumnDataNotAvailable + + err := impl.ProcessMessage(t.Context(), nil, envelope) + require.ErrorIs(t, err, ErrIgnore) + require.Equal(t, int32(1), impl.pendingCount.Load()) + + impl.pendingEnvelopes.Range(func(_, value any) bool { + value.(*envelopeJob).nextAttempt = time.Time{} + return true + }) + impl.processPendingEnvelopes(t.Context()) + require.Equal(t, int32(1), impl.pendingCount.Load()) + impl.pendingEnvelopes.Range(func(_, value any) bool { + job := value.(*envelopeJob) + require.Equal(t, 2*time.Second, job.retryDelay) + require.True(t, job.nextAttempt.After(time.Now())) + return true + }) + + fcu.OnExecutionPayloadErr = nil + impl.pendingEnvelopes.Range(func(_, value any) bool { + value.(*envelopeJob).nextAttempt = time.Time{} + return true + }) + impl.processPendingEnvelopes(t.Context()) + require.Equal(t, int32(0), impl.pendingCount.Load()) + require.True(t, impl.seenEnvelopesCache.Contains(seenEnvelopeKey{blockRoot, 1})) +} + +func TestExecutionPayloadServiceRetriesRecoveredEnvelope(t *testing.T) { + impl, fcu := setupExecutionPayloadServiceWithoutLoop(t) + blockRoot := common.HexToHash("0x1234") + envelope := newTestSignedEnvelope(100, blockRoot, 1) + fcu.Blocks[blockRoot] = newTestSignedBlockWithBuilder(blockRoot, 100, 1) + fcu.OnExecutionPayloadErr = forkchoice.ErrEIP7594ColumnDataNotAvailable + + err := impl.ProcessRecoveredEnvelope(t.Context(), envelope, true) + require.ErrorIs(t, err, ErrIgnore) + require.ErrorIs(t, err, forkchoice.ErrEIP7594ColumnDataNotAvailable) + require.Equal(t, int32(1), impl.pendingCount.Load()) + + fcu.OnExecutionPayloadErr = nil + impl.pendingEnvelopes.Range(func(_, value any) bool { + value.(*envelopeJob).nextAttempt = time.Time{} + return true + }) + impl.processPendingEnvelopes(t.Context()) + require.Equal(t, int32(0), impl.pendingCount.Load()) + require.True(t, impl.seenEnvelopesCache.Contains(seenEnvelopeKey{blockRoot, 1})) +} + +func TestExecutionPayloadServiceDataAvailabilityRetriesDeduplicateByRoot(t *testing.T) { + impl, fcu := setupExecutionPayloadServiceWithoutLoop(t) + blockRoot := common.HexToHash("0x1234") + fcu.Blocks[blockRoot] = newTestSignedBlockWithBuilder(blockRoot, 100, 1) + fcu.OnExecutionPayloadErr = forkchoice.ErrEIP7594ColumnDataNotAvailable + + require.Error(t, impl.ProcessRecoveredEnvelope(t.Context(), newTestSignedEnvelope(100, blockRoot, 1), true)) + require.Error(t, impl.ProcessRecoveredEnvelope(t.Context(), newTestSignedEnvelope(100, blockRoot, 2), true)) + require.Equal(t, int32(1), impl.pendingCount.Load()) +} + +func TestExecutionPayloadServiceFreshDataAvailabilityRetryRefreshesExpiry(t *testing.T) { + impl, _ := setupExecutionPayloadServiceWithoutLoop(t) + blockRoot := common.HexToHash("0x1234") + envelope := newTestSignedEnvelope(100, blockRoot, 1) + impl.queuePendingEnvelopeWithOptions(blockRoot, envelope, false, true, true) + + key := pendingEnvelopeKey{blockRoot: blockRoot, dataAvailability: true} + value, ok := impl.pendingEnvelopes.Load(key) + require.True(t, ok) + value.(*envelopeJob).creationTime = time.Now().Add(-pendingDataAvailabilityExpiry - time.Second) + + impl.queuePendingEnvelopeWithOptions(blockRoot, envelope, true, true, true) + impl.processPendingEnvelopes(t.Context()) + + value, ok = impl.pendingEnvelopes.Load(key) + require.True(t, ok) + require.True(t, value.(*envelopeJob).recovered.Load()) + require.Equal(t, int32(1), impl.pendingCount.Load()) +} + +func TestExecutionPayloadServicePendingExpiryCoversDeferredColumnSync(t *testing.T) { + require.Equal(t, 30*time.Second, pendingEnvelopeExpiry) + require.Greater(t, pendingDataAvailabilityExpiry, time.Minute) +} + +func TestExecutionPayloadServiceRejectsGossipAfterRecoveredEnvelope(t *testing.T) { + impl, fcu := setupExecutionPayloadServiceWithoutLoop(t) + blockRoot := common.HexToHash("0x1234") + valid := newTestSignedEnvelope(100, blockRoot, 1) + fcu.Blocks[blockRoot] = newTestSignedBlockWithBuilder(blockRoot, 100, 1) + require.NoError(t, impl.ProcessRecoveredEnvelope(t.Context(), valid, true)) + fcu.Envelopes[blockRoot] = valid + + forged := newTestSignedEnvelope(100, blockRoot, 2) + err := impl.ProcessMessage(t.Context(), nil, forged) + require.Error(t, err) + require.NotErrorIs(t, err, ErrIgnore) + require.False(t, impl.seenEnvelopesCache.Contains(seenEnvelopeKey{blockRoot, 2})) +} + +func TestExecutionPayloadServiceRejectsConcurrentAlreadyStoredGossip(t *testing.T) { + impl, fcu := setupExecutionPayloadServiceWithoutLoop(t) + blockRoot := common.HexToHash("0x1234") + fcu.Blocks[blockRoot] = newTestSignedBlockWithBuilder(blockRoot, 100, 1) + fcu.Envelopes[blockRoot] = newTestSignedEnvelope(100, blockRoot, 1) + hasEnvelope := false + fcu.HasEnvelopeOverride = &hasEnvelope + fcu.OnExecutionPayloadErr = forkchoice.ErrExecutionPayloadAlreadyStored + + err := impl.ProcessMessage(t.Context(), nil, newTestSignedEnvelope(100, blockRoot, 2)) + require.Error(t, err) + require.NotErrorIs(t, err, ErrIgnore) + require.Equal(t, int32(0), impl.pendingCount.Load()) + require.False(t, impl.seenEnvelopesCache.Contains(seenEnvelopeKey{blockRoot, 2})) +} + +func TestExecutionPayloadServiceIgnoresSameBuilderWithoutStoredEnvelopeRead(t *testing.T) { + impl, fcu := setupExecutionPayloadServiceWithoutLoop(t) + blockRoot := common.HexToHash("0x1234") + fcu.Blocks[blockRoot] = newTestSignedBlockWithBuilder(blockRoot, 100, 1) + fcu.Envelopes[blockRoot] = newTestSignedEnvelope(100, blockRoot, 1) + + err := impl.ProcessMessage(t.Context(), nil, newTestSignedEnvelope(100, blockRoot, 1)) + require.ErrorIs(t, err, ErrIgnore) + require.Zero(t, fcu.ReadEnvelopeCalls.Load()) +} + +func TestExecutionPayloadServiceUsesCommittedBuilderForDuplicateGossip(t *testing.T) { + impl, fcu := setupExecutionPayloadServiceWithoutLoop(t) + blockRoot := common.HexToHash("0x1234") + fcu.Blocks[blockRoot] = newTestSignedBlockWithBuilder(blockRoot, 100, 1) + fcu.Envelopes[blockRoot] = newTestSignedEnvelope(100, blockRoot, 1) + + for builderIndex := uint64(2); builderIndex <= 3; builderIndex++ { + err := impl.ProcessMessage(t.Context(), nil, newTestSignedEnvelope(100, blockRoot, builderIndex)) + require.Error(t, err) + require.NotErrorIs(t, err, ErrIgnore) + } + require.Zero(t, fcu.ReadEnvelopeCalls.Load()) +} + +func TestExecutionPayloadServiceFinalizedDuplicateAvoidsStoredEnvelopeRead(t *testing.T) { + impl, fcu := setupExecutionPayloadServiceWithoutLoop(t) + blockRoot := common.HexToHash("0x1234") + fcu.Blocks[blockRoot] = newTestSignedBlockWithBuilder(blockRoot, 99, 1) + fcu.Envelopes[blockRoot] = newTestSignedEnvelope(99, blockRoot, 1) + fcu.FinalizedSlotVal = 100 + + err := impl.ProcessMessage(t.Context(), nil, newTestSignedEnvelope(99, blockRoot, 2)) + require.ErrorIs(t, err, ErrIgnore) + require.Zero(t, fcu.ReadEnvelopeCalls.Load()) +} + +func TestExecutionPayloadServiceEmitsAvailabilityOnceAcrossProvenance(t *testing.T) { + for _, tt := range []struct { + name string + firstRecovered bool + }{ + {name: "recovery then gossip", firstRecovered: true}, + {name: "gossip then recovery"}, + } { + t.Run(tt.name, func(t *testing.T) { + impl, fcu := setupExecutionPayloadServiceWithoutLoop(t) + blockRoot := common.HexToHash("0x1234") + envelope := newTestSignedEnvelope(100, blockRoot, 1) + fcu.Blocks[blockRoot] = &cltypes.SignedBeaconBlock{Block: &cltypes.BeaconBlock{Slot: 100}} + + events := make(chan *beaconevents.EventStream, 2) + sub := impl.emitters.Operation().Subscribe(events) + defer sub.Unsubscribe() + + if tt.firstRecovered { + require.NoError(t, impl.ProcessRecoveredEnvelope(t.Context(), envelope, true)) + } else { + require.NoError(t, impl.ProcessMessage(t.Context(), nil, envelope)) + } + fcu.Envelopes[blockRoot] = envelope + fcu.OnExecutionPayloadErr = forkchoice.ErrExecutionPayloadAlreadyStored + if tt.firstRecovered { + require.ErrorIs(t, impl.ProcessMessage(t.Context(), nil, envelope), ErrIgnore) + } else { + require.ErrorIs(t, impl.ProcessRecoveredEnvelope(t.Context(), envelope, true), ErrIgnore) + } + + event := <-events + require.Equal(t, beaconevents.OpExecutionPayloadAvailable, event.Event) + select { + case duplicate := <-events: + t.Fatalf("unexpected duplicate event: %v", duplicate) + default: + } + }) + } +} + func TestExecutionPayloadServiceMultiplePendingForSameBlock(t *testing.T) { cfg := &clparams.MainnetBeaconConfig forkchoiceMock := mock_services.NewForkChoiceStorageMock(t) @@ -316,14 +571,18 @@ func TestExecutionPayloadServiceMultiplePendingForSameBlock(t *testing.T) { hash2, _ := envelope2.HashSSZ() // Add both as pending - impl.pendingEnvelopes.Store(pendingEnvelopeKey{blockRoot, hash1}, &envelopeJob{ + job1 := &envelopeJob{ envelope: envelope1, creationTime: time.Now(), - }) - impl.pendingEnvelopes.Store(pendingEnvelopeKey{blockRoot, hash2}, &envelopeJob{ + } + job1.validate.Store(true) + impl.pendingEnvelopes.Store(pendingEnvelopeKey{blockRoot, hash1, false}, job1) + job2 := &envelopeJob{ envelope: envelope2, creationTime: time.Now(), - }) + } + job2.validate.Store(true) + impl.pendingEnvelopes.Store(pendingEnvelopeKey{blockRoot, hash2, false}, job2) impl.pendingCount.Store(2) // Add block @@ -365,10 +624,179 @@ func TestExecutionPayloadServicePendingQueueCap(t *testing.T) { require.Equal(t, int32(maxPendingEnvelopes), impl.pendingCount.Load()) envelopeHash, err := envelope.HashSSZ() require.NoError(t, err) - _, exists := impl.pendingEnvelopes.Load(pendingEnvelopeKey{blockRoot, envelopeHash}) + _, exists := impl.pendingEnvelopes.Load(pendingEnvelopeKey{blockRoot, envelopeHash, false}) require.False(t, exists) } +func TestExecutionPayloadServicePendingQueueCapRejectsUntrustedBeforeHashing(t *testing.T) { + impl, _ := setupExecutionPayloadServiceWithoutLoop(t) + impl.pendingCount.Store(maxPendingEnvelopes) + + require.NotPanics(t, func() { + impl.queuePendingEnvelope(common.HexToHash("0x1234"), &cltypes.SignedExecutionPayloadEnvelope{}) + }) + require.Equal(t, int32(maxPendingEnvelopes), impl.pendingCount.Load()) +} + +func TestExecutionPayloadServicePendingQueueUpgradesDataAvailabilityDuplicateAtCap(t *testing.T) { + impl, _ := setupExecutionPayloadServiceWithoutLoop(t) + blockRoot := common.HexToHash("0x1234") + envelope := newTestSignedEnvelope(100, blockRoot, 1) + impl.queuePendingEnvelopeWithOptions(blockRoot, envelope, false, true, true) + impl.pendingCount.Store(maxPendingEnvelopes) + + impl.queuePendingEnvelopeWithOptions(blockRoot, envelope, true, true, true) + + value, ok := impl.pendingEnvelopes.Load(pendingEnvelopeKey{blockRoot: blockRoot, dataAvailability: true}) + require.True(t, ok) + require.True(t, value.(*envelopeJob).recovered.Load()) + require.Equal(t, int32(maxPendingEnvelopes), impl.pendingCount.Load()) +} + +func TestExecutionPayloadServicePromotesValidatedRetryAtCap(t *testing.T) { + impl, _ := setupExecutionPayloadServiceWithoutLoop(t) + blockRoot := common.HexToHash("0x1234") + envelope := newTestSignedEnvelope(100, blockRoot, 1) + envelopeHash, err := envelope.HashSSZ() + require.NoError(t, err) + oldKey := pendingEnvelopeKey{blockRoot: blockRoot, envelopeHash: envelopeHash} + job := &envelopeJob{envelope: envelope, creationTime: time.Now()} + job.validate.Store(true) + impl.pendingEnvelopes.Store(oldKey, job) + impl.pendingCount.Store(maxPendingEnvelopes) + + impl.promoteDataAvailabilityRetry(oldKey, job) + + _, oldExists := impl.pendingEnvelopes.Load(oldKey) + value, newExists := impl.pendingEnvelopes.Load(pendingEnvelopeKey{blockRoot: blockRoot, dataAvailability: true}) + require.False(t, oldExists) + require.True(t, newExists) + require.Same(t, job, value) + require.Equal(t, int32(maxPendingEnvelopes), impl.pendingCount.Load()) +} + +func TestExecutionPayloadServiceDataAvailabilityRetryEvictsUnvalidatedAtCap(t *testing.T) { + impl, _ := setupExecutionPayloadServiceWithoutLoop(t) + unvalidatedRoot := common.HexToHash("0x1111") + unvalidated := newTestSignedEnvelope(100, unvalidatedRoot, 1) + unvalidatedHash, err := unvalidated.HashSSZ() + require.NoError(t, err) + unvalidatedKey := pendingEnvelopeKey{blockRoot: unvalidatedRoot, envelopeHash: unvalidatedHash} + impl.pendingEnvelopes.Store(unvalidatedKey, &envelopeJob{envelope: unvalidated, creationTime: time.Now()}) + impl.pendingCount.Store(maxPendingEnvelopes) + + validatedRoot := common.HexToHash("0x2222") + validated := newTestSignedEnvelope(100, validatedRoot, 2) + impl.queuePendingEnvelopeWithOptions(validatedRoot, validated, true, true, true) + + _, unvalidatedExists := impl.pendingEnvelopes.Load(unvalidatedKey) + _, validatedExists := impl.pendingEnvelopes.Load(pendingEnvelopeKey{blockRoot: validatedRoot, dataAvailability: true}) + require.False(t, unvalidatedExists) + require.True(t, validatedExists) + require.Equal(t, int32(maxPendingEnvelopes), impl.pendingCount.Load()) +} + +func TestExecutionPayloadServiceDataAvailabilityRetryDoesNotEvictProcessingJob(t *testing.T) { + impl, _ := setupExecutionPayloadServiceWithoutLoop(t) + processingRoot := common.HexToHash("0x1111") + processingEnvelope := newTestSignedEnvelope(100, processingRoot, 1) + processingHash, err := processingEnvelope.HashSSZ() + require.NoError(t, err) + processingKey := pendingEnvelopeKey{blockRoot: processingRoot, envelopeHash: processingHash} + impl.pendingEnvelopes.Store(processingKey, &envelopeJob{envelope: processingEnvelope, creationTime: time.Now(), processing: true}) + + impl.pendingCount.Store(1) + impl.pendingMu.Lock() + impl.evictUnvalidatedPendingEnvelope() + impl.pendingMu.Unlock() + + _, processingExists := impl.pendingEnvelopes.Load(processingKey) + require.True(t, processingExists) + require.Equal(t, int32(1), impl.pendingCount.Load()) +} + +func TestExecutionPayloadServiceInFlightPromotionSurvivesPriorityEviction(t *testing.T) { + impl, fcu := setupExecutionPayloadServiceWithoutLoop(t) + processingRoot := common.HexToHash("0x1111") + processingEnvelope := newTestSignedEnvelope(100, processingRoot, 1) + processingHash, err := processingEnvelope.HashSSZ() + require.NoError(t, err) + processingKey := pendingEnvelopeKey{blockRoot: processingRoot, envelopeHash: processingHash} + processingJob := &envelopeJob{envelope: processingEnvelope, creationTime: time.Now()} + processingJob.validate.Store(true) + impl.pendingEnvelopes.Store(processingKey, processingJob) + fcu.Blocks[processingRoot] = newTestSignedBlockWithBuilder(processingRoot, 100, 1) + + evictableRoot := common.HexToHash("0x2222") + evictableEnvelope := newTestSignedEnvelope(100, evictableRoot, 2) + evictableHash, err := evictableEnvelope.HashSSZ() + require.NoError(t, err) + impl.pendingEnvelopes.Store( + pendingEnvelopeKey{blockRoot: evictableRoot, envelopeHash: evictableHash}, + &envelopeJob{envelope: evictableEnvelope, creationTime: time.Now()}, + ) + impl.pendingCount.Store(maxPendingEnvelopes) + + entered := make(chan struct{}) + release := make(chan struct{}) + fcu.OnExecutionPayloadFunc = func(context.Context, *cltypes.SignedExecutionPayloadEnvelope, bool, bool) error { + close(entered) + <-release + return forkchoice.ErrEIP7594ColumnDataNotAvailable + } + done := make(chan struct{}) + go func() { + impl.processPendingEnvelopes(t.Context()) + close(done) + }() + <-entered + + priorityRoot := common.HexToHash("0x3333") + impl.queuePendingEnvelopeWithOptions(priorityRoot, newTestSignedEnvelope(100, priorityRoot, 3), true, true, true) + close(release) + <-done + + _, promoted := impl.pendingEnvelopes.Load(pendingEnvelopeKey{blockRoot: processingRoot, dataAvailability: true}) + _, priority := impl.pendingEnvelopes.Load(pendingEnvelopeKey{blockRoot: priorityRoot, dataAvailability: true}) + require.True(t, promoted) + require.True(t, priority) +} + +func TestExecutionPayloadServicePendingDuplicateUpgradesAtCap(t *testing.T) { + impl, _ := setupExecutionPayloadServiceWithoutLoop(t) + blockRoot := common.HexToHash("0x1234") + envelope := newTestSignedEnvelope(100, blockRoot, 1) + impl.queuePendingEnvelopeWithOptions(blockRoot, envelope, false, false, false) + impl.pendingCount.Store(maxPendingEnvelopes) + + impl.queuePendingEnvelopeWithOptions(blockRoot, envelope, true, true, false) + + hash, err := envelope.HashSSZ() + require.NoError(t, err) + value, ok := impl.pendingEnvelopes.Load(pendingEnvelopeKey{blockRoot: blockRoot, envelopeHash: hash}) + require.True(t, ok) + require.True(t, value.(*envelopeJob).recovered.Load()) + require.True(t, value.(*envelopeJob).validate.Load()) +} + +func TestExecutionPayloadServicePendingQueuePreservesRecoveredEnvelope(t *testing.T) { + impl, _ := setupExecutionPayloadServiceWithoutLoop(t) + blockRoot := common.HexToHash("0x1234") + envelope := newTestSignedEnvelope(100, blockRoot, 1) + envelopeHash, err := envelope.HashSSZ() + require.NoError(t, err) + + impl.queuePendingEnvelopeWithOptions(blockRoot, envelope, true, false, false) + impl.queuePendingEnvelope(blockRoot, envelope) + + value, ok := impl.pendingEnvelopes.Load(pendingEnvelopeKey{blockRoot, envelopeHash, false}) + require.True(t, ok) + job := value.(*envelopeJob) + require.True(t, job.recovered.Load()) + require.True(t, job.validate.Load()) + require.Equal(t, int32(1), impl.pendingCount.Load()) +} + func TestExecutionPayloadServicePendingQueueCapConcurrent(t *testing.T) { cfg := &clparams.MainnetBeaconConfig forkchoiceMock := mock_services.NewForkChoiceStorageMock(t) diff --git a/cl/phase1/network/services/mock_services/execution_payload_service_mock.go b/cl/phase1/network/services/mock_services/execution_payload_service_mock.go index e31cd094101..8543d32ef24 100644 --- a/cl/phase1/network/services/mock_services/execution_payload_service_mock.go +++ b/cl/phase1/network/services/mock_services/execution_payload_service_mock.go @@ -157,3 +157,41 @@ func (c *MockExecutionPayloadServiceProcessMessageCall) DoAndReturn(f func(conte c.Call = c.Call.DoAndReturn(f) return c } + +// ProcessRecoveredEnvelope mocks base method. +func (m *MockExecutionPayloadService) ProcessRecoveredEnvelope(arg0 context.Context, arg1 *cltypes.SignedExecutionPayloadEnvelope, arg2 bool) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ProcessRecoveredEnvelope", arg0, arg1, arg2) + ret0, _ := ret[0].(error) + return ret0 +} + +// ProcessRecoveredEnvelope indicates an expected call of ProcessRecoveredEnvelope. +func (mr *MockExecutionPayloadServiceMockRecorder) ProcessRecoveredEnvelope(arg0, arg1, arg2 any) *MockExecutionPayloadServiceProcessRecoveredEnvelopeCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ProcessRecoveredEnvelope", reflect.TypeOf((*MockExecutionPayloadService)(nil).ProcessRecoveredEnvelope), arg0, arg1, arg2) + return &MockExecutionPayloadServiceProcessRecoveredEnvelopeCall{Call: call} +} + +// MockExecutionPayloadServiceProcessRecoveredEnvelopeCall wrap *gomock.Call +type MockExecutionPayloadServiceProcessRecoveredEnvelopeCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *MockExecutionPayloadServiceProcessRecoveredEnvelopeCall) Return(arg0 error) *MockExecutionPayloadServiceProcessRecoveredEnvelopeCall { + c.Call = c.Call.Return(arg0) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *MockExecutionPayloadServiceProcessRecoveredEnvelopeCall) Do(f func(context.Context, *cltypes.SignedExecutionPayloadEnvelope, bool) error) *MockExecutionPayloadServiceProcessRecoveredEnvelopeCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *MockExecutionPayloadServiceProcessRecoveredEnvelopeCall) DoAndReturn(f func(context.Context, *cltypes.SignedExecutionPayloadEnvelope, bool) error) *MockExecutionPayloadServiceProcessRecoveredEnvelopeCall { + c.Call = c.Call.DoAndReturn(f) + return c +} diff --git a/cl/phase1/network/services/types.go b/cl/phase1/network/services/types.go index 1a83733d789..c128add96c4 100644 --- a/cl/phase1/network/services/types.go +++ b/cl/phase1/network/services/types.go @@ -1,6 +1,8 @@ package services import ( + "context" + "github.com/erigontech/erigon/cl/cltypes" serviceinterface "github.com/erigontech/erigon/cl/phase1/network/services/service_interface" ) @@ -39,7 +41,10 @@ type DataColumnSidecarService serviceinterface.Service[*cltypes.DataColumnSideca type AttesterSlashingService serviceinterface.Service[*cltypes.AttesterSlashing] //go:generate mockgen -typed=true -destination=./mock_services/execution_payload_service_mock.go -package=mock_services . ExecutionPayloadService -type ExecutionPayloadService serviceinterface.Service[*cltypes.SignedExecutionPayloadEnvelope] +type ExecutionPayloadService interface { + serviceinterface.Service[*cltypes.SignedExecutionPayloadEnvelope] + ProcessRecoveredEnvelope(context.Context, *cltypes.SignedExecutionPayloadEnvelope, bool) error +} //go:generate mockgen -typed=true -destination=./mock_services/execution_payload_bid_service_mock.go -package=mock_services . ExecutionPayloadBidService type ExecutionPayloadBidService serviceinterface.Service[*cltypes.SignedExecutionPayloadBid] diff --git a/cl/phase1/stages/chain_tip_sync.go b/cl/phase1/stages/chain_tip_sync.go index 02dffddf278..381f08abc88 100644 --- a/cl/phase1/stages/chain_tip_sync.go +++ b/cl/phase1/stages/chain_tip_sync.go @@ -22,7 +22,14 @@ import ( "github.com/erigontech/erigon/common/log/v3" ) -const maxGloasVerificationSweepPerCycle = 32 +const ( + maxGloasVerificationSweepPerCycle = 32 + maxGloasVerificationScanPerLineage = 256 + maxGloasVerificationStartRootsPerCycle = 8 + maxGloasVerificationLeafRootsPerCycle = maxGloasVerificationStartRootsPerCycle - 2 + maxGloasVerificationStalledCycles = 8 + maxGloasVerificationCheckpoints = 256 +) func gloasVersionedHashes(blobCommitments *solid.ListSSZ[*cltypes.KZGCommitment]) ([]common.Hash, error) { if blobCommitments == nil || blobCommitments.Len() == 0 { @@ -64,6 +71,10 @@ func canRetryGloasPayloads(cfg *Cfg) bool { return cfg.executionClient != nil && cfg.executionClient.SupportInsertion() } +func canValidateGloasPayloads(cfg *Cfg) bool { + return cfg.executionClient != nil +} + // waitForExecutionEngineToBeFinished checks if the execution engine is ready within a specified timeout. // It periodically checks the readiness of the execution client and returns true if the client is ready before // the timeout occurs. If the context is canceled or a timeout occurs, it returns false with the corresponding error. @@ -259,7 +270,7 @@ MainLoop: if block.Version() >= clparams.GloasVersion && len(envelopes) > 0 { parentRoot := block.Block.ParentRoot if env, ok := envelopes[common.Hash(parentRoot)]; ok { - if envErr := cfg.forkChoice.OnExecutionPayload(ctx, env, false, canRetryGloasPayloads(cfg)); envErr != nil { + if envErr := cfg.forkChoice.OnExecutionPayload(ctx, env, false, canValidateGloasPayloads(cfg)); envErr != nil && !errors.Is(envErr, forkchoice.ErrExecutionPayloadAlreadyStored) { log.Debug("[chainTipSync] failed to apply parent envelope", "slot", block.Block.Slot, "err", envErr) } } @@ -295,9 +306,23 @@ func fetchAndApplyEnvelopes(ctx context.Context, cfg *Cfg, roots [][32]byte) { log.Debug("[chainTipSync] failed to request GLOAS envelopes", "err", err) return } - for _, env := range envelopes { - if err := cfg.forkChoice.OnExecutionPayload(ctx, env, true, canRetryGloasPayloads(cfg)); err != nil { - log.Debug("[chainTipSync] failed to apply recovered GLOAS envelope", "beaconBlockRoot", env.Message.BeaconBlockRoot, "err", err) + applyRecoveredEnvelopes(ctx, cfg, envelopes) +} + +func applyRecoveredEnvelopes(ctx context.Context, cfg *Cfg, envelopes map[common.Hash]*cltypes.SignedExecutionPayloadEnvelope) { + for root, env := range envelopes { + if env == nil || env.Message == nil { + log.Debug("[chainTipSync] ignoring invalid recovered GLOAS envelope", "beaconBlockRoot", root) + continue + } + var err error + if cfg.recoveredEnvelopeProcessor != nil { + err = cfg.recoveredEnvelopeProcessor.ProcessRecoveredEnvelope(ctx, env, true) + } else { + err = cfg.forkChoice.OnExecutionPayload(ctx, env, true, true) + } + if err != nil && !errors.Is(err, forkchoice.ErrExecutionPayloadAlreadyStored) { + log.Debug("[chainTipSync] failed to apply recovered GLOAS envelope", "beaconBlockRoot", root, "err", err) } } } @@ -547,43 +572,58 @@ func drainPendingGloasPayloads(ctx context.Context, cfg *Cfg) { } func verifyUnverifiedGloasPayloads(ctx context.Context, cfg *Cfg) { - headRoot := cfg.forkChoice.HighestSeenRoot() - if headRoot == (common.Hash{}) { + cfg.gloasVerificationMu.Lock() + if cfg.gloasVerificationRunning { + cfg.gloasVerificationMu.Unlock() return } - - finalizedSlot := cfg.forkChoice.FinalizedSlot() - var blocks []struct { - root common.Hash - block *cltypes.SignedBeaconBlock - } - - for root := headRoot; root != (common.Hash{}); { - block, ok := cfg.forkChoice.GetBlock(root) - if !ok || block == nil { - break - } - if block.Block.Slot <= finalizedSlot { - break - } - epoch := block.Block.Slot / cfg.beaconCfg.SlotsPerEpoch - if cfg.beaconCfg.GetCurrentStateVersion(epoch) < clparams.GloasVersion { - break - } - if cfg.forkChoice.HasEnvelope(root) && !cfg.forkChoice.IsPayloadVerified(root) { - blocks = append(blocks, struct { - root common.Hash - block *cltypes.SignedBeaconBlock - }{root: root, block: block}) - if len(blocks) >= maxGloasVerificationSweepPerCycle { - break - } + cfg.gloasVerificationRunning = true + cfg.gloasVerificationMu.Unlock() + defer func() { + cfg.gloasVerificationMu.Lock() + cfg.gloasVerificationRunning = false + cfg.gloasVerificationMu.Unlock() + }() + + startRoots := make([]common.Hash, 0, 2) + addStartRoot := func(root common.Hash) { + if root == (common.Hash{}) || slices.Contains(startRoots, root) || len(startRoots) >= maxGloasVerificationStartRootsPerCycle { + return } - root = common.Hash(block.Block.ParentRoot) + startRoots = append(startRoots, root) + } + canonicalRoot, _, err := cfg.forkChoice.GetHead(nil) + if err != nil { + log.Warn("[chainTipSync] failed to resolve canonical head for GLOAS verification", "err", err) + } + addStartRoot(canonicalRoot) + addStartRoot(cfg.forkChoice.HighestSeenRoot()) + cfg.gloasVerificationLineages = prioritizeGloasVerificationLineages(cfg.gloasVerificationLineages, startRoots) + leafLimit := min(maxGloasVerificationLeafRootsPerCycle, maxGloasVerificationStartRootsPerCycle-len(cfg.gloasVerificationLineages)) + if leafLimit > 0 { + cfg.gloasVerificationLineages = mergeGloasVerificationLineages( + cfg.gloasVerificationLineages, + cfg.forkChoice.GloasVerificationLeaves(leafLimit), + ) + } + if len(cfg.gloasVerificationLineages) == 0 { + return } + blocks, lineages := collectUnverifiedGloasPayloadPages( + cfg.gloasVerificationLineages, + cfg.forkChoice.FinalizedSlot(), + cfg.beaconCfg, + cfg.forkChoice.GetBlock, + func(root common.Hash) bool { + status, hasStatus := cfg.forkChoice.GetRecentExecutionPayloadStatusByRoot(root) + return shouldVerifyGloasPayload(cfg.forkChoice.HasEnvelope(root), cfg.forkChoice.IsPayloadVerified(root), status, hasStatus) + }, + ) + cfg.gloasVerificationLineages = lineages + swept := 0 - for _, item := range slices.Backward(blocks) { + for _, item := range blocks { if cfg.forkChoice.IsPayloadVerified(item.root) { continue } @@ -621,6 +661,368 @@ func verifyUnverifiedGloasPayloads(ctx context.Context, cfg *Cfg) { } } +type gloasVerificationBlock struct { + root common.Hash + block *cltypes.SignedBeaconBlock +} + +type gloasVerificationLineage struct { + origin common.Hash + cursor common.Hash + readyBoundary common.Hash + checkpoints []common.Hash + pending int + stalled int + truncated bool +} + +func mergeGloasVerificationLineages(states []gloasVerificationLineage, roots []common.Hash) []gloasVerificationLineage { + for _, root := range roots { + if root == (common.Hash{}) || len(states) >= maxGloasVerificationStartRootsPerCycle { + continue + } + known := false + for i := range states { + if states[i].origin == root || states[i].cursor == root { + known = true + break + } + } + if !known { + states = append(states, gloasVerificationLineage{origin: root, cursor: root, checkpoints: []common.Hash{root}}) + } + } + return states +} + +func prioritizeGloasVerificationLineages(states []gloasVerificationLineage, roots []common.Hash) []gloasVerificationLineage { + prioritized := make([]gloasVerificationLineage, 0, maxGloasVerificationStartRootsPerCycle) + selected := make(map[int]struct{}, len(roots)) + for _, root := range roots { + if len(prioritized) >= maxGloasVerificationStartRootsPerCycle { + break + } + if root == (common.Hash{}) { + continue + } + found := -1 + for i := range states { + if states[i].origin == root || states[i].cursor == root { + found = i + break + } + } + if found >= 0 { + if _, ok := selected[found]; !ok { + prioritized = append(prioritized, states[found]) + selected[found] = struct{}{} + } + } else { + prioritized = append(prioritized, gloasVerificationLineage{origin: root, cursor: root, checkpoints: []common.Hash{root}}) + } + } + for i := range states { + if len(prioritized) >= maxGloasVerificationStartRootsPerCycle { + break + } + if _, ok := selected[i]; !ok { + prioritized = append(prioritized, states[i]) + } + } + return prioritized +} + +func shouldVerifyGloasPayload(hasEnvelope, verified bool, status execution_client.PayloadStatus, hasStatus bool) bool { + if !hasEnvelope || verified { + return false + } + if !hasStatus { + return true + } + return status != execution_client.PayloadStatusValidated && status != execution_client.PayloadStatusInvalidated +} + +func gloasPageDependsOnBoundary(page []gloasVerificationBlock, boundary *cltypes.SignedBeaconBlock) bool { + if boundary == nil || boundary.Block == nil { + return true + } + boundaryBid := boundary.Block.Body.GetSignedExecutionPayloadBid() + if boundaryBid == nil || boundaryBid.Message == nil { + return true + } + for _, item := range page { + if item.block == nil || item.block.Block == nil { + return true + } + bid := item.block.Block.Body.GetSignedExecutionPayloadBid() + if bid == nil || bid.Message == nil || bid.Message.ParentBlockHash == boundaryBid.Message.BlockHash { + return true + } + } + return false +} + +func gloasPageDependsOnSharedAncestry( + page []gloasVerificationBlock, + root common.Hash, + getBlock func(common.Hash) (*cltypes.SignedBeaconBlock, bool), + shouldVerify func(common.Hash) bool, +) bool { + for scanned := 0; root != (common.Hash{}) && scanned < maxGloasVerificationScanPerLineage; scanned++ { + block, ok := getBlock(root) + if !ok || block == nil || block.Block == nil { + return true + } + if shouldVerify(root) && gloasPageDependsOnBoundary(page, block) { + return true + } + root = common.Hash(block.Block.ParentRoot) + } + return root != (common.Hash{}) +} + +func collectUnverifiedGloasPayloadPages( + states []gloasVerificationLineage, + finalizedSlot uint64, + beaconCfg *clparams.BeaconChainConfig, + getBlock func(common.Hash) (*cltypes.SignedBeaconBlock, bool), + shouldVerify func(common.Hash) bool, +) ([]gloasVerificationBlock, []gloasVerificationLineage) { + return collectUnverifiedGloasPayloadPagesWithCheckpointLimit( + states, + finalizedSlot, + beaconCfg, + getBlock, + shouldVerify, + maxGloasVerificationCheckpoints, + ) +} + +func collectUnverifiedGloasPayloadPagesWithCheckpointLimit( + states []gloasVerificationLineage, + finalizedSlot uint64, + beaconCfg *clparams.BeaconChainConfig, + getBlock func(common.Hash) (*cltypes.SignedBeaconBlock, bool), + shouldVerify func(common.Hash) bool, + checkpointLimit int, +) ([]gloasVerificationBlock, []gloasVerificationLineage) { + seen := make(map[common.Hash]struct{}, maxGloasVerificationScanPerLineage) + pages := make([][]gloasVerificationBlock, 0, len(states)) + next := make([]gloasVerificationLineage, 0, len(states)) + for i := range states { + state := states[i] + if len(state.checkpoints) == 0 { + state.checkpoints = []common.Hash{state.origin} + } + page := make([]gloasVerificationBlock, 0) + pageRoots := make([]common.Hash, 0, maxGloasVerificationScanPerLineage) + root := state.cursor + scanned := 0 + blocked := false + unavailable := false + for ; root != (common.Hash{}) && scanned < maxGloasVerificationScanPerLineage; scanned++ { + if _, ok := seen[root]; ok { + if gloasPageDependsOnSharedAncestry(page, root, getBlock, shouldVerify) { + blocked = true + } + root = common.Hash{} + break + } + seen[root] = struct{}{} + pageRoots = append(pageRoots, root) + block, ok := getBlock(root) + if !ok || block == nil || block.Block == nil { + unavailable = true + root = common.Hash{} + break + } + if block.Block.Slot <= finalizedSlot { + root = common.Hash{} + break + } + epoch := block.Block.Slot / beaconCfg.SlotsPerEpoch + if beaconCfg.GetCurrentStateVersion(epoch) < clparams.GloasVersion { + root = common.Hash{} + break + } + if shouldVerify(root) { + page = append(page, gloasVerificationBlock{root: root, block: block}) + } + root = common.Hash(block.Block.ParentRoot) + } + if blocked { + for _, pageRoot := range pageRoots { + delete(seen, pageRoot) + } + next = append(next, state) + continue + } + if unavailable { + state.pending = 0 + state.stalled++ + if state.stalled < maxGloasVerificationStalledCycles { + next = append(next, state) + } + continue + } + if scanned == maxGloasVerificationScanPerLineage && root != (common.Hash{}) && root != state.readyBoundary { + if gloasPageDependsOnSharedAncestry(page, root, getBlock, shouldVerify) { + for _, pageRoot := range pageRoots { + delete(seen, pageRoot) + } + page = page[:0] + } + state.cursor = root + if state.checkpoints[len(state.checkpoints)-1] != root { + state.checkpoints = append(state.checkpoints, root) + if len(state.checkpoints) > checkpointLimit { + state.checkpoints = append([]common.Hash(nil), state.checkpoints[len(state.checkpoints)-checkpointLimit:]...) + state.truncated = true + } + } + state.pending = 0 + state.stalled = 0 + next = append(next, state) + slices.Reverse(page) + if len(page) > 0 { + pages = append(pages, page) + } + continue + } + slices.Reverse(page) + if len(page) > 0 { + pages = append(pages, page) + if state.pending == len(page) { + state.stalled++ + } else { + state.pending = len(page) + state.stalled = 0 + } + if state.stalled < maxGloasVerificationStalledCycles { + next = append(next, state) + } + continue + } + if len(state.checkpoints) > 1 { + state.readyBoundary = state.cursor + state.checkpoints = state.checkpoints[:len(state.checkpoints)-1] + state.cursor = state.checkpoints[len(state.checkpoints)-1] + state.pending = 0 + state.stalled = 0 + next = append(next, state) + } else if state.truncated { + state.readyBoundary = state.cursor + state.cursor = state.origin + state.checkpoints = []common.Hash{state.origin} + state.pending = 0 + state.stalled = 0 + state.truncated = false + next = append(next, state) + } + } + return interleaveGloasVerificationPages(pages), next +} + +func interleaveGloasVerificationPages(pages [][]gloasVerificationBlock) []gloasVerificationBlock { + blocks := make([]gloasVerificationBlock, 0, maxGloasVerificationSweepPerCycle) + for index := 0; len(blocks) < maxGloasVerificationSweepPerCycle; index++ { + added := false + for _, page := range pages { + if index >= len(page) { + continue + } + blocks = append(blocks, page[index]) + added = true + if len(blocks) >= maxGloasVerificationSweepPerCycle { + break + } + } + if !added { + break + } + } + slices.SortStableFunc(blocks, func(a, b gloasVerificationBlock) int { + return cmp.Compare(a.block.Block.Slot, b.block.Block.Slot) + }) + return blocks +} + +func collectUnverifiedGloasPayloads( + startRoots []common.Hash, + finalizedSlot uint64, + beaconCfg *clparams.BeaconChainConfig, + getBlock func(common.Hash) (*cltypes.SignedBeaconBlock, bool), + shouldVerify func(common.Hash) bool, +) ([]gloasVerificationBlock, []common.Hash) { + blocks := make([]gloasVerificationBlock, 0, maxGloasVerificationSweepPerCycle) + seen := make(map[common.Hash]struct{}, maxGloasVerificationScanPerLineage) + lineages := make([][]gloasVerificationBlock, 0, min(len(startRoots), maxGloasVerificationStartRootsPerCycle)) + continuations := make([]common.Hash, 0, len(lineages)) + for _, startRoot := range startRoots { + if len(lineages) >= maxGloasVerificationStartRootsPerCycle { + break + } + lineage := make([]gloasVerificationBlock, 0) + lineageRoots := make([]common.Hash, 0, maxGloasVerificationScanPerLineage) + root := startRoot + scanned := 0 + for ; root != (common.Hash{}) && scanned < maxGloasVerificationScanPerLineage; scanned++ { + if _, ok := seen[root]; ok { + if shouldVerify(root) { + lineage = lineage[:0] + } + root = common.Hash{} + break + } + seen[root] = struct{}{} + lineageRoots = append(lineageRoots, root) + block, ok := getBlock(root) + if !ok || block == nil || block.Block == nil || block.Block.Slot <= finalizedSlot { + break + } + epoch := block.Block.Slot / beaconCfg.SlotsPerEpoch + if beaconCfg.GetCurrentStateVersion(epoch) < clparams.GloasVersion { + break + } + if shouldVerify(root) { + lineage = append(lineage, gloasVerificationBlock{root: root, block: block}) + } + root = common.Hash(block.Block.ParentRoot) + } + if scanned == maxGloasVerificationScanPerLineage && root != (common.Hash{}) { + lineage = lineage[:0] + for _, lineageRoot := range lineageRoots { + delete(seen, lineageRoot) + } + if !slices.Contains(continuations, root) { + continuations = append(continuations, root) + } + } + slices.Reverse(lineage) + lineages = append(lineages, lineage) + } + for index := 0; len(blocks) < maxGloasVerificationSweepPerCycle; index++ { + added := false + for _, lineage := range lineages { + if index >= len(lineage) { + continue + } + blocks = append(blocks, lineage[index]) + added = true + if len(blocks) >= maxGloasVerificationSweepPerCycle { + break + } + } + if !added { + break + } + } + slices.SortStableFunc(blocks, func(a, b gloasVerificationBlock) int { + return cmp.Compare(a.block.Block.Slot, b.block.Block.Slot) + }) + return blocks, continuations +} + func retryUnverifiedAnchorPayload(ctx context.Context, cfg *Cfg) { anchorSlot := cfg.forkChoice.AnchorSlot() epoch := anchorSlot / cfg.beaconCfg.SlotsPerEpoch @@ -676,11 +1078,13 @@ func chainTipSync(ctx context.Context, logger log.Logger, cfg *Cfg, args Args) e // insertion — so it must run regardless of SupportInsertion(). recoverMissingEnvelopes(ctx, cfg) - if canRetryGloasPayloads(cfg) { + if canValidateGloasPayloads(cfg) { // [New in Gloas:EIP7732] Drain execution blocks whose CL transition succeeded // but whose EL newPayload previously returned SYNCING/ACCEPTED. drainPendingGloasPayloads(ctx, cfg) retryUnverifiedAnchorPayload(ctx, cfg) + } + if canRetryGloasPayloads(cfg) { if err := cfg.blockCollector.Flush(context.Background()); err != nil { log.Warn("[chainTipSync] blockCollector.Flush failed (EL may still be catching up)", "err", err) } @@ -696,7 +1100,7 @@ func chainTipSync(ctx context.Context, logger log.Logger, cfg *Cfg, args Args) e if headRoot != (common.Hash{}) && !cfg.forkChoice.HasEnvelope(headRoot) { pollForEnvelope(ctx, cfg, headRoot, 2*time.Second) } - if canRetryGloasPayloads(cfg) { + if canValidateGloasPayloads(cfg) { verifyUnverifiedGloasPayloads(ctx, cfg) } // NOTE: recoverMissingEnvelopes runs unconditionally above (before diff --git a/cl/phase1/stages/chain_tip_sync_test.go b/cl/phase1/stages/chain_tip_sync_test.go new file mode 100644 index 00000000000..3c7b05e156b --- /dev/null +++ b/cl/phase1/stages/chain_tip_sync_test.go @@ -0,0 +1,65 @@ +// Copyright 2026 The Erigon Authors +// This file is part of Erigon. +// +// Erigon is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// Erigon is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with Erigon. If not, see . + +package stages + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/erigontech/erigon/cl/clparams" + "github.com/erigontech/erigon/cl/cltypes" + "github.com/erigontech/erigon/common" +) + +type recoveredEnvelopeProcessorStub struct { + envelopes []*cltypes.SignedExecutionPayloadEnvelope + validate []bool +} + +func (s *recoveredEnvelopeProcessorStub) ProcessRecoveredEnvelope(_ context.Context, envelope *cltypes.SignedExecutionPayloadEnvelope, validate bool) error { + s.envelopes = append(s.envelopes, envelope) + s.validate = append(s.validate, validate) + return nil +} + +func TestApplyRecoveredEnvelopesUsesRetryingProcessor(t *testing.T) { + processor := &recoveredEnvelopeProcessorStub{} + envelope := &cltypes.SignedExecutionPayloadEnvelope{ + Message: cltypes.NewExecutionPayloadEnvelope(&clparams.MainnetBeaconConfig), + } + cfg := &Cfg{recoveredEnvelopeProcessor: processor} + + applyRecoveredEnvelopes(t.Context(), cfg, map[common.Hash]*cltypes.SignedExecutionPayloadEnvelope{ + common.HexToHash("0x1234"): envelope, + }) + + require.Equal(t, []*cltypes.SignedExecutionPayloadEnvelope{envelope}, processor.envelopes) + require.Equal(t, []bool{true}, processor.validate) +} + +func TestApplyRecoveredEnvelopesIgnoresNilEnvelope(t *testing.T) { + processor := &recoveredEnvelopeProcessorStub{} + cfg := &Cfg{recoveredEnvelopeProcessor: processor} + + applyRecoveredEnvelopes(t.Context(), cfg, map[common.Hash]*cltypes.SignedExecutionPayloadEnvelope{ + common.HexToHash("0x1234"): nil, + }) + + require.Empty(t, processor.envelopes) +} diff --git a/cl/phase1/stages/clstages.go b/cl/phase1/stages/clstages.go index 84746b66d00..b5980b3dbb6 100644 --- a/cl/phase1/stages/clstages.go +++ b/cl/phase1/stages/clstages.go @@ -20,6 +20,7 @@ import ( "context" "errors" "fmt" + "sync" "time" "github.com/erigontech/erigon/cl/antiquary" @@ -48,26 +49,34 @@ import ( ) type Cfg struct { - rpc *rpc.BeaconRpcP2P - ethClock eth_clock.EthereumClock - beaconCfg *clparams.BeaconChainConfig - executionClient execution_client.ExecutionEngine - state *state.CachingBeaconState - forkChoice *forkchoice.ForkChoiceStore - indiciesDB kv.RwDB - dirs datadir.Dirs - blockReader freezeblocks.BeaconSnapshotReader - antiquary *antiquary.Antiquary - syncedData *synced_data.SyncedDataManager - emitter *beaconevents.EventEmitter - blockCollector block_collector.BlockCollector - sn *freezeblocks.CaplinSnapshots - blobStore blob_storage.BlobStorage - peerDas das.PeerDas - blobDownloader *network2.BlobHistoryDownloader - attestationDataProducer attestation_producer.AttestationDataProducer - caplinConfig clparams.CaplinConfig - hasDownloaded bool + rpc *rpc.BeaconRpcP2P + ethClock eth_clock.EthereumClock + beaconCfg *clparams.BeaconChainConfig + executionClient execution_client.ExecutionEngine + state *state.CachingBeaconState + forkChoice *forkchoice.ForkChoiceStore + recoveredEnvelopeProcessor recoveredEnvelopeProcessor + indiciesDB kv.RwDB + dirs datadir.Dirs + blockReader freezeblocks.BeaconSnapshotReader + antiquary *antiquary.Antiquary + syncedData *synced_data.SyncedDataManager + emitter *beaconevents.EventEmitter + blockCollector block_collector.BlockCollector + sn *freezeblocks.CaplinSnapshots + blobStore blob_storage.BlobStorage + peerDas das.PeerDas + blobDownloader *network2.BlobHistoryDownloader + attestationDataProducer attestation_producer.AttestationDataProducer + caplinConfig clparams.CaplinConfig + hasDownloaded bool + gloasVerificationMu sync.Mutex + gloasVerificationRunning bool + gloasVerificationLineages []gloasVerificationLineage +} + +type recoveredEnvelopeProcessor interface { + ProcessRecoveredEnvelope(context.Context, *cltypes.SignedExecutionPayloadEnvelope, bool) error } type Args struct { @@ -88,6 +97,7 @@ func ClStagesCfg( state *state.CachingBeaconState, executionClient execution_client.ExecutionEngine, forkChoice *forkchoice.ForkChoiceStore, + recoveredEnvelopeProcessor recoveredEnvelopeProcessor, indiciesDB kv.RwDB, sn *freezeblocks.CaplinSnapshots, blockReader freezeblocks.BeaconSnapshotReader, @@ -115,25 +125,26 @@ func ClStagesCfg( ) return &Cfg{ - rpc: rpc, - antiquary: antiquary, - ethClock: ethClock, - caplinConfig: caplinConfig, - beaconCfg: beaconCfg, - state: state, - executionClient: executionClient, - forkChoice: forkChoice, - dirs: dirs, - indiciesDB: indiciesDB, - sn: sn, - blockReader: blockReader, - peerDas: peerDas, - blobDownloader: blobDownloader, - syncedData: syncedData, - emitter: emitters, - blobStore: blobStore, - blockCollector: block_collector.NewPersistentBlockCollector(log.Root(), executionClient, beaconCfg, dirs.CaplinHistory), - attestationDataProducer: attestationDataProducer, + rpc: rpc, + antiquary: antiquary, + ethClock: ethClock, + caplinConfig: caplinConfig, + beaconCfg: beaconCfg, + state: state, + executionClient: executionClient, + forkChoice: forkChoice, + recoveredEnvelopeProcessor: recoveredEnvelopeProcessor, + dirs: dirs, + indiciesDB: indiciesDB, + sn: sn, + blockReader: blockReader, + peerDas: peerDas, + blobDownloader: blobDownloader, + syncedData: syncedData, + emitter: emitters, + blobStore: blobStore, + blockCollector: block_collector.NewPersistentBlockCollector(log.Root(), executionClient, beaconCfg, dirs.CaplinHistory), + attestationDataProducer: attestationDataProducer, } } diff --git a/cl/phase1/stages/forward_sync.go b/cl/phase1/stages/forward_sync.go index 9c59c469597..0032b61a354 100644 --- a/cl/phase1/stages/forward_sync.go +++ b/cl/phase1/stages/forward_sync.go @@ -210,7 +210,7 @@ func processDownloadedBlockBatches(ctx context.Context, logger log.Logger, cfg * if block.Version() >= clparams.GloasVersion { if env, ok := envelopes[blockRoot]; ok { // FULL block: update forkchoice with the envelope (updates eth2Roots, persists to disk). - if fceErr := cfg.forkChoice.OnExecutionPayload(ctx, env, false, false); fceErr != nil { + if fceErr := cfg.forkChoice.OnExecutionPayload(ctx, env, false, canValidateGloasPayloads(cfg)); fceErr != nil && !errors.Is(fceErr, forkchoice.ErrExecutionPayloadAlreadyStored) { logger.Warn("[Caplin] forward sync: failed to process GLOAS envelope", "slot", block.Block.Slot, "err", fceErr) } else if shouldInsert { if err = cfg.blockCollector.AddGloasBlock(block.Block, env); err != nil { @@ -360,9 +360,16 @@ func forwardSync(ctx context.Context, logger log.Logger, cfg *Cfg, args Args) er } lastProgressTime := time.Now() lastProgressSlot := currentSlot.Load() + lastGloasVerification := time.Time{} // Run the log loop until the highest processed slot reaches the chain tip slot for downloader.GetHighestProcessedSlot() < chainTipSlot { + if canValidateGloasPayloads(cfg) && time.Since(lastGloasVerification) >= time.Second { + drainPendingGloasPayloads(ctx, cfg) + retryUnverifiedAnchorPayload(ctx, cfg) + verifyUnverifiedGloasPayloads(ctx, cfg) + lastGloasVerification = time.Now() + } downloader.RequestMore(ctx) // Detect stale progress: if no new slots processed for staleTimeout, exit. @@ -515,7 +522,7 @@ func ensureAnchorEnvelopeOnce(ctx context.Context, cfg *Cfg) error { if err := cfg.forkChoice.StoreAnchorEnvelope(anchorRoot, env); err != nil { return fmt.Errorf("failed to store anchor envelope: %w", err) } - if err := validateAnchorPayloadIfLocalEL(ctx, cfg, anchorRoot, bid, env); err != nil { + if err := validateAnchorPayloadWithExecutionClient(ctx, cfg, anchorRoot, bid, env); err != nil { return err } @@ -524,8 +531,8 @@ func ensureAnchorEnvelopeOnce(ctx context.Context, cfg *Cfg) error { return nil } -func validateAnchorPayloadIfLocalEL(ctx context.Context, cfg *Cfg, anchorRoot common.Hash, bid *cltypes.ExecutionPayloadBid, env *cltypes.SignedExecutionPayloadEnvelope) error { - if !canRetryGloasPayloads(cfg) { +func validateAnchorPayloadWithExecutionClient(ctx context.Context, cfg *Cfg, anchorRoot common.Hash, bid *cltypes.ExecutionPayloadBid, env *cltypes.SignedExecutionPayloadEnvelope) error { + if !canValidateGloasPayloads(cfg) { return nil } status, err := validateAnchorPayloadWithEL(ctx, cfg, bid, env) diff --git a/cl/phase1/stages/gloas_payload_test.go b/cl/phase1/stages/gloas_payload_test.go index 46f1bd88b74..4383f6f36c4 100644 --- a/cl/phase1/stages/gloas_payload_test.go +++ b/cl/phase1/stages/gloas_payload_test.go @@ -2,7 +2,9 @@ package stages import ( "context" + "encoding/binary" "math/big" + "slices" "testing" "github.com/holiman/uint256" @@ -124,6 +126,591 @@ func TestValidateAnchorEnvelope(t *testing.T) { } } +func TestCollectUnverifiedGloasPayloadsIncludesCanonicalAndHighestSeenForks(t *testing.T) { + cfg := clparams.MainnetBeaconConfig + clparams.ApplyMinimalPreset(&cfg) + cfg.AltairForkEpoch = 0 + cfg.BellatrixForkEpoch = 0 + cfg.CapellaForkEpoch = 0 + cfg.DenebForkEpoch = 0 + cfg.ElectraForkEpoch = 0 + cfg.FuluForkEpoch = 0 + cfg.GloasForkEpoch = 0 + cfg.InitializeForkSchedule() + parentRoot := common.HexToHash("0x10") + rootA := common.HexToHash("0xa") + rootB := common.HexToHash("0xb") + blocks := map[common.Hash]*cltypes.SignedBeaconBlock{ + parentRoot: testGloasVerificationBlock(&cfg, 1, common.HexToHash("0x01")), + rootA: testGloasVerificationBlock(&cfg, 2, parentRoot), + rootB: testGloasVerificationBlock(&cfg, 2, parentRoot), + } + getBlock := func(root common.Hash) (*cltypes.SignedBeaconBlock, bool) { + block, ok := blocks[root] + return block, ok + } + shouldVerify := func(root common.Hash) bool { + _, ok := blocks[root] + return ok + } + + for _, tt := range []struct { + name string + starts []common.Hash + }{ + {name: "canonical B and highest-seen A", starts: []common.Hash{rootB, rootA}}, + {name: "canonical A and highest-seen B", starts: []common.Hash{rootA, rootB}}, + } { + t.Run(tt.name, func(t *testing.T) { + items, _ := collectUnverifiedGloasPayloads(tt.starts, 0, &cfg, getBlock, shouldVerify) + roots := verificationRoots(items) + require.Len(t, roots, 2) + require.Equal(t, parentRoot, roots[0]) + require.Equal(t, tt.starts[0], roots[1]) + }) + } +} + +func TestCollectUnverifiedGloasPayloadsDoesNotStarveLineageAtScanLimit(t *testing.T) { + cfg := testGloasVerificationConfig() + blocks := make(map[common.Hash]*cltypes.SignedBeaconBlock, maxGloasVerificationScanPerLineage+2) + canonicalRoot := common.Hash{} + for i := 1; i <= maxGloasVerificationScanPerLineage+1; i++ { + root := testGloasVerificationRoot(uint64(i)) + blocks[root] = testGloasVerificationBlock(&cfg, uint64(i), canonicalRoot) + canonicalRoot = root + } + sideRoot := testGloasVerificationRoot(10_000) + blocks[sideRoot] = testGloasVerificationBlock(&cfg, 1, common.Hash{}) + getBlock := func(root common.Hash) (*cltypes.SignedBeaconBlock, bool) { + block, ok := blocks[root] + return block, ok + } + + for _, length := range []int{maxGloasVerificationScanPerLineage, maxGloasVerificationScanPerLineage + 1} { + canonicalRoot = testGloasVerificationRoot(uint64(length)) + for _, starts := range [][]common.Hash{{canonicalRoot, sideRoot}, {sideRoot, canonicalRoot}} { + items, _ := collectUnverifiedGloasPayloads(starts, 0, &cfg, getBlock, func(root common.Hash) bool { + return root == sideRoot + }) + require.Contains(t, verificationRoots(items), sideRoot) + } + } +} + +func TestCollectUnverifiedGloasPayloadsReturnsDeepLineageContinuation(t *testing.T) { + cfg := testGloasVerificationConfig() + blocks := make(map[common.Hash]*cltypes.SignedBeaconBlock, maxGloasVerificationScanPerLineage+1) + parentRoot := common.Hash{} + oldestRoot := common.Hash{} + for i := 1; i <= maxGloasVerificationScanPerLineage+1; i++ { + root := testGloasVerificationRoot(uint64(i)) + blocks[root] = testGloasVerificationBlock(&cfg, uint64(i), parentRoot) + if i == 1 { + oldestRoot = root + } + parentRoot = root + } + newestRoot := parentRoot + verified := make(map[common.Hash]bool) + shouldVerify := func(root common.Hash) bool { return !verified[root] } + items, continuations := collectUnverifiedGloasPayloads( + []common.Hash{newestRoot}, + 0, + &cfg, + func(root common.Hash) (*cltypes.SignedBeaconBlock, bool) { + block, ok := blocks[root] + return block, ok + }, + shouldVerify, + ) + + require.Empty(t, items) + require.Equal(t, []common.Hash{oldestRoot}, continuations) + items, continuations = collectUnverifiedGloasPayloads( + continuations, + 0, + &cfg, + func(root common.Hash) (*cltypes.SignedBeaconBlock, bool) { + block, ok := blocks[root] + return block, ok + }, + shouldVerify, + ) + require.Empty(t, continuations) + require.Equal(t, []common.Hash{oldestRoot}, verificationRoots(items)) +} + +func TestCollectUnverifiedGloasPayloadsContinuesThroughEmptyBoundaryWithoutDeferringDescendant(t *testing.T) { + cfg := testGloasVerificationConfig() + blocks := make(map[common.Hash]*cltypes.SignedBeaconBlock, maxGloasVerificationScanPerLineage+1) + parentRoot := common.Hash{} + for i := 1; i <= maxGloasVerificationScanPerLineage+1; i++ { + root := testGloasVerificationRoot(uint64(i)) + blocks[root] = testGloasVerificationBlock(&cfg, uint64(i), parentRoot) + parentRoot = root + } + newestRoot := parentRoot + items, continuations := collectUnverifiedGloasPayloads( + []common.Hash{newestRoot}, + 0, + &cfg, + func(root common.Hash) (*cltypes.SignedBeaconBlock, bool) { + block, ok := blocks[root] + return block, ok + }, + func(root common.Hash) bool { return root == newestRoot }, + ) + + require.Equal(t, []common.Hash{testGloasVerificationRoot(1)}, continuations) + require.Empty(t, items) +} + +func TestCollectUnverifiedGloasPayloadsFindsWorkBehindNonActionableBoundary(t *testing.T) { + cfg := testGloasVerificationConfig() + blocks := make(map[common.Hash]*cltypes.SignedBeaconBlock, maxGloasVerificationScanPerLineage+2) + parentRoot := common.Hash{} + for i := 1; i <= maxGloasVerificationScanPerLineage+2; i++ { + root := testGloasVerificationRoot(uint64(i)) + blocks[root] = testGloasVerificationBlock(&cfg, uint64(i), parentRoot) + parentRoot = root + } + oldestRoot := testGloasVerificationRoot(1) + boundaryRoot := testGloasVerificationRoot(2) + blocks[oldestRoot].Block.Body.GetSignedExecutionPayloadBid().Message.BlockHash = common.HexToHash("0x01") + blocks[boundaryRoot].Block.Body.GetSignedExecutionPayloadBid().Message.ParentBlockHash = common.HexToHash("0x02") + getBlock := func(root common.Hash) (*cltypes.SignedBeaconBlock, bool) { + block, ok := blocks[root] + return block, ok + } + shouldVerify := func(root common.Hash) bool { return root == oldestRoot } + + items, continuations := collectUnverifiedGloasPayloads([]common.Hash{parentRoot}, 0, &cfg, getBlock, shouldVerify) + require.Empty(t, items) + require.Equal(t, []common.Hash{boundaryRoot}, continuations) + items, continuations = collectUnverifiedGloasPayloads(continuations, 0, &cfg, getBlock, shouldVerify) + require.Empty(t, continuations) + require.Equal(t, []common.Hash{oldestRoot}, verificationRoots(items)) +} + +func TestCollectUnverifiedGloasPayloadsDoesNotDeferIndependentEmptyDescendants(t *testing.T) { + cfg := testGloasVerificationConfig() + blocks := make(map[common.Hash]*cltypes.SignedBeaconBlock, maxGloasVerificationScanPerLineage+1) + parentRoot := common.Hash{} + for i := 1; i <= maxGloasVerificationScanPerLineage+1; i++ { + root := testGloasVerificationRoot(uint64(i)) + blocks[root] = testGloasVerificationBlock(&cfg, uint64(i), parentRoot) + parentRoot = root + } + oldestRoot := testGloasVerificationRoot(1) + oldestChildRoot := testGloasVerificationRoot(2) + blocks[oldestRoot].Block.Body.GetSignedExecutionPayloadBid().Message.BlockHash = common.HexToHash("0x01") + blocks[oldestChildRoot].Block.Body.GetSignedExecutionPayloadBid().Message.ParentBlockHash = common.HexToHash("0x02") + items, continuations := collectUnverifiedGloasPayloads( + []common.Hash{parentRoot}, + 0, + &cfg, + func(root common.Hash) (*cltypes.SignedBeaconBlock, bool) { + block, ok := blocks[root] + return block, ok + }, + func(common.Hash) bool { return true }, + ) + + require.Equal(t, []common.Hash{oldestRoot}, continuations) + require.Empty(t, items) +} + +func TestCollectUnverifiedGloasPayloadsDefersSharedForkUntilAncestor(t *testing.T) { + cfg := testGloasVerificationConfig() + blocks := make(map[common.Hash]*cltypes.SignedBeaconBlock, maxGloasVerificationScanPerLineage+2) + parentRoot := common.Hash{} + for i := 1; i <= maxGloasVerificationScanPerLineage+1; i++ { + root := testGloasVerificationRoot(uint64(i)) + blocks[root] = testGloasVerificationBlock(&cfg, uint64(i), parentRoot) + parentRoot = root + } + oldestRoot := testGloasVerificationRoot(1) + sharedRoot := testGloasVerificationRoot(200) + sideRoot := testGloasVerificationRoot(10_000) + blocks[sideRoot] = testGloasVerificationBlock(&cfg, maxGloasVerificationScanPerLineage+2, sharedRoot) + items, continuations := collectUnverifiedGloasPayloads( + []common.Hash{parentRoot, sideRoot}, + 0, + &cfg, + func(root common.Hash) (*cltypes.SignedBeaconBlock, bool) { + block, ok := blocks[root] + return block, ok + }, + func(common.Hash) bool { return true }, + ) + + require.Equal(t, []common.Hash{oldestRoot}, continuations) + require.NotEmpty(t, items) + require.Equal(t, oldestRoot, items[0].root) + require.NotContains(t, verificationRoots(items), sideRoot) +} + +func TestCollectUnverifiedGloasPayloadPagesAdvancesBeyondContinuationCapacity(t *testing.T) { + cfg := testGloasVerificationConfig() + checkpointLimit := maxGloasVerificationStartRootsPerCycle + depth := maxGloasVerificationScanPerLineage*(checkpointLimit*3) + 1 + blocks := make(map[common.Hash]*cltypes.SignedBeaconBlock, depth) + parentRoot := common.Hash{} + for i := 1; i <= depth; i++ { + root := testGloasVerificationRoot(uint64(i)) + blocks[root] = testGloasVerificationBlock(&cfg, uint64(i), parentRoot) + parentRoot = root + } + verified := make(map[common.Hash]bool, depth) + states := []gloasVerificationLineage{{origin: parentRoot, cursor: parentRoot}} + getBlockCalls := 0 + for cycle := 0; cycle < 500 && !verified[parentRoot]; cycle++ { + if len(states) == 0 { + states = []gloasVerificationLineage{{origin: parentRoot, cursor: parentRoot}} + } + items, next := collectUnverifiedGloasPayloadPagesWithCheckpointLimit( + states, + 0, + &cfg, + func(root common.Hash) (*cltypes.SignedBeaconBlock, bool) { + getBlockCalls++ + block, ok := blocks[root] + return block, ok + }, + func(root common.Hash) bool { return !verified[root] }, + checkpointLimit, + ) + for _, item := range items { + verified[item.root] = true + } + states = next + for _, state := range states { + require.LessOrEqual(t, len(state.checkpoints), checkpointLimit) + } + } + + require.True(t, verified[testGloasVerificationRoot(1)]) + require.True(t, verified[parentRoot]) + require.Less(t, getBlockCalls, depth*20) +} + +func TestCollectUnverifiedGloasPayloadPagesDefersSharedSideUntilParent(t *testing.T) { + cfg := testGloasVerificationConfig() + blocks := make(map[common.Hash]*cltypes.SignedBeaconBlock, 66) + parentRoot := common.Hash{} + for i := 1; i <= 64; i++ { + root := testGloasVerificationRoot(uint64(i)) + blocks[root] = testGloasVerificationBlock(&cfg, uint64(i), parentRoot) + parentRoot = root + } + sharedRoot := parentRoot + canonicalRoot := testGloasVerificationRoot(65) + sideRoot := testGloasVerificationRoot(10_000) + blocks[canonicalRoot] = testGloasVerificationBlock(&cfg, 65, sharedRoot) + blocks[sideRoot] = testGloasVerificationBlock(&cfg, 65, sharedRoot) + verified := make(map[common.Hash]bool) + states := []gloasVerificationLineage{ + {origin: canonicalRoot, cursor: canonicalRoot}, + {origin: sideRoot, cursor: sideRoot}, + } + sawSide := false + for cycle := 0; cycle < 20 && len(states) > 0; cycle++ { + items, next := collectUnverifiedGloasPayloadPages( + states, + 0, + &cfg, + func(root common.Hash) (*cltypes.SignedBeaconBlock, bool) { + block, ok := blocks[root] + return block, ok + }, + func(root common.Hash) bool { return !verified[root] }, + ) + for _, item := range items { + if item.root == sideRoot { + require.True(t, verified[sharedRoot]) + sawSide = true + } + verified[item.root] = true + } + states = next + } + require.True(t, sawSide) +} + +func TestCollectUnverifiedGloasPayloadPagesProcessesIndependentEmptyBoundary(t *testing.T) { + cfg := testGloasVerificationConfig() + blocks := make(map[common.Hash]*cltypes.SignedBeaconBlock, maxGloasVerificationScanPerLineage+1) + parentRoot := common.Hash{} + for i := 1; i <= maxGloasVerificationScanPerLineage+1; i++ { + root := testGloasVerificationRoot(uint64(i)) + blocks[root] = testGloasVerificationBlock(&cfg, uint64(i), parentRoot) + parentRoot = root + } + boundaryRoot := testGloasVerificationRoot(1) + blocks[boundaryRoot].Block.Body.GetSignedExecutionPayloadBid().Message.BlockHash = common.HexToHash("0x01") + newestRoot := parentRoot + items, next := collectUnverifiedGloasPayloadPages( + []gloasVerificationLineage{{origin: newestRoot, cursor: newestRoot}}, + 0, + &cfg, + func(root common.Hash) (*cltypes.SignedBeaconBlock, bool) { + block, ok := blocks[root] + return block, ok + }, + func(root common.Hash) bool { return root == newestRoot || root == boundaryRoot }, + ) + + require.Equal(t, []common.Hash{newestRoot}, verificationRoots(items)) + require.NotEmpty(t, next) +} + +func TestCollectUnverifiedGloasPayloadPagesDefersTransitiveFullBoundary(t *testing.T) { + cfg := testGloasVerificationConfig() + blocks := make(map[common.Hash]*cltypes.SignedBeaconBlock, maxGloasVerificationScanPerLineage+2) + parentRoot := common.Hash{} + for i := 1; i <= maxGloasVerificationScanPerLineage+2; i++ { + root := testGloasVerificationRoot(uint64(i)) + blocks[root] = testGloasVerificationBlock(&cfg, uint64(i), parentRoot) + parentRoot = root + } + fullAncestorRoot := testGloasVerificationRoot(1) + fullAncestorHash := common.HexToHash("0x01") + blocks[fullAncestorRoot].Block.Body.GetSignedExecutionPayloadBid().Message.BlockHash = fullAncestorHash + boundaryRoot := testGloasVerificationRoot(2) + blocks[boundaryRoot].Block.Body.GetSignedExecutionPayloadBid().Message.ParentBlockHash = fullAncestorHash + oldestPageRoot := testGloasVerificationRoot(3) + blocks[oldestPageRoot].Block.Body.GetSignedExecutionPayloadBid().Message.ParentBlockHash = fullAncestorHash + newestRoot := parentRoot + items, next := collectUnverifiedGloasPayloadPages( + []gloasVerificationLineage{{origin: newestRoot, cursor: newestRoot}}, + 0, + &cfg, + func(root common.Hash) (*cltypes.SignedBeaconBlock, bool) { + block, ok := blocks[root] + return block, ok + }, + func(root common.Hash) bool { return root == oldestPageRoot || root == fullAncestorRoot }, + ) + + require.Empty(t, items) + require.Equal(t, boundaryRoot, next[0].cursor) +} + +func TestCollectUnverifiedGloasPayloadPagesDefersSharedEmptyUntilFullAncestor(t *testing.T) { + cfg := testGloasVerificationConfig() + blocks := make(map[common.Hash]*cltypes.SignedBeaconBlock, 66) + parentRoot := common.Hash{} + for i := 1; i <= 64; i++ { + root := testGloasVerificationRoot(uint64(i)) + blocks[root] = testGloasVerificationBlock(&cfg, uint64(i), parentRoot) + parentRoot = root + } + sharedEmptyRoot := parentRoot + fullAncestorRoot := testGloasVerificationRoot(63) + fullAncestorHash := common.HexToHash("0x63") + blocks[fullAncestorRoot].Block.Body.GetSignedExecutionPayloadBid().Message.BlockHash = fullAncestorHash + canonicalRoot := testGloasVerificationRoot(65) + sideRoot := testGloasVerificationRoot(10_000) + blocks[canonicalRoot] = testGloasVerificationBlock(&cfg, 65, sharedEmptyRoot) + blocks[sideRoot] = testGloasVerificationBlock(&cfg, 65, sharedEmptyRoot) + blocks[sideRoot].Block.Body.GetSignedExecutionPayloadBid().Message.ParentBlockHash = fullAncestorHash + items, _ := collectUnverifiedGloasPayloadPages( + []gloasVerificationLineage{ + {origin: canonicalRoot, cursor: canonicalRoot}, + {origin: sideRoot, cursor: sideRoot}, + }, + 0, + &cfg, + func(root common.Hash) (*cltypes.SignedBeaconBlock, bool) { + block, ok := blocks[root] + return block, ok + }, + func(root common.Hash) bool { return root != sharedEmptyRoot }, + ) + + require.NotContains(t, verificationRoots(items), sideRoot) +} + +func TestShouldVerifyGloasPayloadExcludesTerminalStatus(t *testing.T) { + require.True(t, shouldVerifyGloasPayload(true, false, execution_client.PayloadStatusNone, false)) + require.True(t, shouldVerifyGloasPayload(true, false, execution_client.PayloadStatusNotValidated, true)) + require.False(t, shouldVerifyGloasPayload(true, false, execution_client.PayloadStatusInvalidated, true)) + require.False(t, shouldVerifyGloasPayload(true, false, execution_client.PayloadStatusValidated, true)) + require.False(t, shouldVerifyGloasPayload(true, true, execution_client.PayloadStatusValidated, true)) + require.False(t, shouldVerifyGloasPayload(false, false, execution_client.PayloadStatusNone, false)) +} + +func TestMergeGloasVerificationLineagesAddsNewRootWhileActive(t *testing.T) { + activeRoot := testGloasVerificationRoot(1) + newRoot := testGloasVerificationRoot(2) + states := mergeGloasVerificationLineages( + []gloasVerificationLineage{{origin: activeRoot, cursor: activeRoot}}, + []common.Hash{activeRoot, newRoot}, + ) + + require.Len(t, states, 2) + require.Equal(t, newRoot, states[1].origin) +} + +func TestPrioritizeGloasVerificationLineagesReplacesOldRootAtCapacity(t *testing.T) { + states := make([]gloasVerificationLineage, 0, maxGloasVerificationStartRootsPerCycle) + for i := 1; i <= maxGloasVerificationStartRootsPerCycle; i++ { + root := testGloasVerificationRoot(uint64(i)) + states = append(states, gloasVerificationLineage{origin: root, cursor: root}) + } + currentRoot := testGloasVerificationRoot(100) + states = prioritizeGloasVerificationLineages(states, []common.Hash{currentRoot}) + + require.Len(t, states, maxGloasVerificationStartRootsPerCycle) + require.Equal(t, currentRoot, states[0].origin) +} + +func TestCollectUnverifiedGloasPayloadPagesDropsStalledLineage(t *testing.T) { + cfg := testGloasVerificationConfig() + root := testGloasVerificationRoot(1) + block := testGloasVerificationBlock(&cfg, 1, common.Hash{}) + states := []gloasVerificationLineage{{origin: root, cursor: root}} + for cycle := 0; cycle <= maxGloasVerificationStalledCycles && len(states) > 0; cycle++ { + _, states = collectUnverifiedGloasPayloadPages( + states, + 0, + &cfg, + func(common.Hash) (*cltypes.SignedBeaconBlock, bool) { return block, true }, + func(common.Hash) bool { return true }, + ) + } + + require.Empty(t, states) +} + +func TestCollectUnverifiedGloasPayloadPagesDoesNotPromoteMissingBoundary(t *testing.T) { + cfg := testGloasVerificationConfig() + blocks := make(map[common.Hash]*cltypes.SignedBeaconBlock, maxGloasVerificationScanPerLineage) + parentRoot := common.Hash{} + for i := 2; i <= maxGloasVerificationScanPerLineage+1; i++ { + root := testGloasVerificationRoot(uint64(i)) + blocks[root] = testGloasVerificationBlock(&cfg, uint64(i), parentRoot) + parentRoot = root + } + missingRoot := testGloasVerificationRoot(1) + blocks[testGloasVerificationRoot(2)].Block.ParentRoot = missingRoot + states := []gloasVerificationLineage{{origin: parentRoot, cursor: parentRoot}} + getBlock := func(root common.Hash) (*cltypes.SignedBeaconBlock, bool) { + block, ok := blocks[root] + return block, ok + } + items, states := collectUnverifiedGloasPayloadPages(states, 0, &cfg, getBlock, func(common.Hash) bool { return true }) + require.Empty(t, items) + require.Equal(t, missingRoot, states[0].cursor) + + items, states = collectUnverifiedGloasPayloadPages(states, 0, &cfg, getBlock, func(common.Hash) bool { return true }) + require.Empty(t, items) + require.Equal(t, missingRoot, states[0].cursor) + require.Equal(t, common.Hash{}, states[0].readyBoundary) +} + +func TestCollectUnverifiedGloasPayloadPagesSharesOutputAcrossLineages(t *testing.T) { + cfg := testGloasVerificationConfig() + blocks := make(map[common.Hash]*cltypes.SignedBeaconBlock, maxGloasVerificationSweepPerCycle+1) + canonicalRoot := common.Hash{} + for i := 1; i <= maxGloasVerificationSweepPerCycle; i++ { + root := testGloasVerificationRoot(uint64(i)) + blocks[root] = testGloasVerificationBlock(&cfg, uint64(i), canonicalRoot) + canonicalRoot = root + } + sideRoot := testGloasVerificationRoot(10_000) + blocks[sideRoot] = testGloasVerificationBlock(&cfg, 1, common.Hash{}) + getBlock := func(root common.Hash) (*cltypes.SignedBeaconBlock, bool) { + block, ok := blocks[root] + return block, ok + } + + for _, starts := range [][]common.Hash{{canonicalRoot, sideRoot}, {sideRoot, canonicalRoot}} { + states := make([]gloasVerificationLineage, 0, len(starts)) + for _, root := range starts { + states = append(states, gloasVerificationLineage{origin: root, cursor: root}) + } + items, _ := collectUnverifiedGloasPayloadPages(states, 0, &cfg, getBlock, func(common.Hash) bool { return true }) + require.LessOrEqual(t, len(items), maxGloasVerificationSweepPerCycle) + require.Contains(t, verificationRoots(items), sideRoot) + lastSlot := uint64(0) + for _, item := range items { + require.GreaterOrEqual(t, item.block.Block.Slot, lastSlot) + lastSlot = item.block.Block.Slot + } + } +} + +func TestCollectUnverifiedGloasPayloadPagesWalksHiddenLeafLineage(t *testing.T) { + cfg := testGloasVerificationConfig() + canonicalRoot := testGloasVerificationRoot(100) + highestSeenRoot := testGloasVerificationRoot(200) + hiddenRoots := []common.Hash{ + testGloasVerificationRoot(301), + testGloasVerificationRoot(302), + testGloasVerificationRoot(303), + } + blocks := map[common.Hash]*cltypes.SignedBeaconBlock{ + canonicalRoot: testGloasVerificationBlock(&cfg, 4, common.Hash{}), + highestSeenRoot: testGloasVerificationBlock(&cfg, 5, common.Hash{}), + hiddenRoots[0]: testGloasVerificationBlock(&cfg, 1, common.Hash{}), + hiddenRoots[1]: testGloasVerificationBlock(&cfg, 2, hiddenRoots[0]), + hiddenRoots[2]: testGloasVerificationBlock(&cfg, 3, hiddenRoots[1]), + } + items, _ := collectUnverifiedGloasPayloadPages( + []gloasVerificationLineage{ + {origin: canonicalRoot, cursor: canonicalRoot}, + {origin: highestSeenRoot, cursor: highestSeenRoot}, + {origin: hiddenRoots[2], cursor: hiddenRoots[2]}, + }, + 0, + &cfg, + func(root common.Hash) (*cltypes.SignedBeaconBlock, bool) { + block, ok := blocks[root] + return block, ok + }, + func(root common.Hash) bool { return slices.Contains(hiddenRoots, root) }, + ) + + require.Equal(t, hiddenRoots, verificationRoots(items)) +} + +func testGloasVerificationConfig() clparams.BeaconChainConfig { + cfg := clparams.MainnetBeaconConfig + clparams.ApplyMinimalPreset(&cfg) + cfg.AltairForkEpoch = 0 + cfg.BellatrixForkEpoch = 0 + cfg.CapellaForkEpoch = 0 + cfg.DenebForkEpoch = 0 + cfg.ElectraForkEpoch = 0 + cfg.FuluForkEpoch = 0 + cfg.GloasForkEpoch = 0 + cfg.InitializeForkSchedule() + return cfg +} + +func testGloasVerificationRoot(i uint64) common.Hash { + var root common.Hash + binary.BigEndian.PutUint64(root[len(root)-8:], i) + return root +} + +func verificationRoots(items []gloasVerificationBlock) []common.Hash { + roots := make([]common.Hash, len(items)) + for i, item := range items { + roots[i] = item.root + } + return roots +} + +func testGloasVerificationBlock(cfg *clparams.BeaconChainConfig, slot uint64, parentRoot common.Hash) *cltypes.SignedBeaconBlock { + block := cltypes.NewSignedBeaconBlock(cfg, clparams.GloasVersion) + block.Block.Slot = slot + block.Block.ParentRoot = parentRoot + return block +} + func TestAnchorEnvelopeMatches(t *testing.T) { _, _, _, env, anchorRoot := validAnchorEnvelopeFixture(t, 1) @@ -187,25 +774,31 @@ func TestStandaloneExecutionClientDoesNotRunLocalGloasRetry(t *testing.T) { require.True(t, canRetryGloasPayloads(&Cfg{executionClient: &testExecutionEngine{supportInsertion: true}})) } -func TestValidateAnchorPayloadIfLocalELFollowsSupportInsertion(t *testing.T) { +func TestStandaloneExecutionClientCanValidateGloasPayloads(t *testing.T) { + require.False(t, canValidateGloasPayloads(&Cfg{})) + require.True(t, canValidateGloasPayloads(&Cfg{executionClient: &testExecutionEngine{supportInsertion: false}})) + require.True(t, canValidateGloasPayloads(&Cfg{executionClient: &testExecutionEngine{supportInsertion: true}})) +} + +func TestValidateAnchorPayloadUsesRemoteExecutionClient(t *testing.T) { cfg, _, bid, env, anchorRoot := validAnchorEnvelopeFixture(t, 1) remoteEL := &testExecutionEngine{ supportInsertion: false, - payloadStatus: execution_client.PayloadStatusInvalidated, + payloadStatus: execution_client.PayloadStatusValidated, } - require.NoError(t, validateAnchorPayloadIfLocalEL(context.Background(), &Cfg{ + require.NoError(t, validateAnchorPayloadWithExecutionClient(context.Background(), &Cfg{ beaconCfg: cfg, executionClient: remoteEL, forkChoice: &forkchoice.ForkChoiceStore{}, }, anchorRoot, bid, env)) - require.Equal(t, 0, remoteEL.newPayloadCalls) + require.Equal(t, 1, remoteEL.newPayloadCalls) localEL := &testExecutionEngine{ supportInsertion: true, payloadStatus: execution_client.PayloadStatusValidated, } - require.NoError(t, validateAnchorPayloadIfLocalEL(context.Background(), &Cfg{ + require.NoError(t, validateAnchorPayloadWithExecutionClient(context.Background(), &Cfg{ beaconCfg: cfg, executionClient: localEL, forkChoice: &forkchoice.ForkChoiceStore{}, @@ -213,6 +806,42 @@ func TestValidateAnchorPayloadIfLocalELFollowsSupportInsertion(t *testing.T) { require.Equal(t, 1, localEL.newPayloadCalls) } +func TestValidateAnchorPayloadHandlesRemoteExecutionStatuses(t *testing.T) { + for _, tt := range []struct { + name string + status execution_client.PayloadStatus + }{ + {name: "validated", status: execution_client.PayloadStatusValidated}, + {name: "syncing", status: execution_client.PayloadStatusNotValidated}, + } { + t.Run(tt.name, func(t *testing.T) { + cfg, _, bid, env, anchorRoot := validAnchorEnvelopeFixture(t, 1) + engine := &testExecutionEngine{payloadStatus: tt.status} + err := validateAnchorPayloadWithExecutionClient(context.Background(), &Cfg{ + beaconCfg: cfg, + executionClient: engine, + forkChoice: &forkchoice.ForkChoiceStore{}, + }, anchorRoot, bid, env) + require.NoError(t, err) + require.Equal(t, 1, engine.newPayloadCalls) + }) + } +} + +func TestValidateAnchorPayloadWithELReturnsRemoteInvalidation(t *testing.T) { + cfg, _, bid, env, _ := validAnchorEnvelopeFixture(t, 1) + engine := &testExecutionEngine{payloadStatus: execution_client.PayloadStatusInvalidated} + + status, err := validateAnchorPayloadWithEL(context.Background(), &Cfg{ + beaconCfg: cfg, + executionClient: engine, + }, bid, env) + + require.NoError(t, err) + require.EqualValues(t, execution_client.PayloadStatusInvalidated, status) + require.Equal(t, 1, engine.newPayloadCalls) +} + func TestDrainPendingGloasPayloadsRequeuesNotValidatedPayload(t *testing.T) { cfg := clparams.MainnetBeaconConfig clparams.ApplyMinimalPreset(&cfg) diff --git a/cl/spectest/consensus_tests/fork_choice.go b/cl/spectest/consensus_tests/fork_choice.go index 34371ac8fdb..9f8b17309e4 100644 --- a/cl/spectest/consensus_tests/fork_choice.go +++ b/cl/spectest/consensus_tests/fork_choice.go @@ -18,6 +18,7 @@ package consensus_tests import ( "context" + "errors" "fmt" "io/fs" "math" @@ -423,6 +424,9 @@ func (b *ForkChoice) Run(t *testing.T, root fs.FS, c spectest.TestCase) (err err err := spectest.ReadSsz(root, c.Version(), step.GetExecutionPayload()+".ssz_snappy", envelope) require.NoError(t, err, stepstr) err = forkStore.OnExecutionPayload(ctx, envelope, false, true) + if errors.Is(err, forkchoice.ErrExecutionPayloadAlreadyStored) { + err = nil + } if step.GetValid() { require.NoError(t, err, stepstr) } else { diff --git a/cmd/caplin/caplin1/run.go b/cmd/caplin/caplin1/run.go index a744357bc8e..b203b776b31 100644 --- a/cmd/caplin/caplin1/run.go +++ b/cmd/caplin/caplin1/run.go @@ -82,6 +82,19 @@ import ( p2pnat "github.com/erigontech/erigon/p2p/nat" ) +// ErrGloasExecutionEngineUnavailable indicates that a scheduled Gloas fork has no execution engine. +var ErrGloasExecutionEngineUnavailable = errors.New("Gloas requires an execution engine") + +func validateGloasExecutionEngine(beaconConfig *clparams.BeaconChainConfig, hasExecutionEngine bool) error { + if beaconConfig == nil { + return fmt.Errorf("%w: missing beacon configuration", ErrGloasExecutionEngineUnavailable) + } + if !hasExecutionEngine && beaconConfig.GloasForkEpoch != math.MaxUint64 { + return ErrGloasExecutionEngineUnavailable + } + return nil +} + func OpenCaplinDatabase(ctx context.Context, beaconConfig *clparams.BeaconChainConfig, ethClock eth_clock.EthereumClock, @@ -253,6 +266,9 @@ func RunCaplinService(ctx context.Context, engine execution_client.ExecutionEngi } } } + if err := validateGloasExecutionEngine(beaconConfig, engine != nil); err != nil { + return err + } // init the current beacon config for global access clparams.InitGlobalStaticConfig(beaconConfig, &config) @@ -639,6 +655,7 @@ func RunCaplinService(ctx context.Context, engine execution_client.ExecutionEngi engine, //gossipManager, forkChoice, + executionPayloadService, indexDB, csn, rcsn, diff --git a/cmd/caplin/caplin1/run_test.go b/cmd/caplin/caplin1/run_test.go new file mode 100644 index 00000000000..c4f7392964e --- /dev/null +++ b/cmd/caplin/caplin1/run_test.go @@ -0,0 +1,34 @@ +package caplin1 + +import ( + "math" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/erigontech/erigon/cl/clparams" +) + +func TestValidateGloasExecutionEngine(t *testing.T) { + t.Run("missing beacon configuration", func(t *testing.T) { + require.ErrorIs(t, validateGloasExecutionEngine(nil, true), ErrGloasExecutionEngineUnavailable) + }) + + t.Run("scheduled without execution engine", func(t *testing.T) { + cfg := clparams.MainnetBeaconConfig + cfg.GloasForkEpoch = 0 + require.ErrorIs(t, validateGloasExecutionEngine(&cfg, false), ErrGloasExecutionEngineUnavailable) + }) + + t.Run("scheduled with execution engine", func(t *testing.T) { + cfg := clparams.MainnetBeaconConfig + cfg.GloasForkEpoch = 0 + require.NoError(t, validateGloasExecutionEngine(&cfg, true)) + }) + + t.Run("unscheduled without execution engine", func(t *testing.T) { + cfg := clparams.MainnetBeaconConfig + cfg.GloasForkEpoch = math.MaxUint64 + require.NoError(t, validateGloasExecutionEngine(&cfg, false)) + }) +}