Merge branch 'develop' into el-offline-api-change

* rollback on SaveState error

* add test

CHANGELOG.md

@@ -23,6 +23,9 @@ The format is based on Keep a Changelog, and this project adheres to Semantic Ve
 - Fix `engine_exchangeCapabilities` implementation.
 - Updated the default `scrape-interval` in `Client-stats` to 2 minutes to accommodate Beaconcha.in API rate limits.
 - Switch to compounding when consolidating with source==target.
+- Revert block db save when saving state fails.
+- Return false from HasBlock if the block is being synced. 
+- Cleanup forkchoice on failed insertions.
 
 ### Deprecated
 
@@ -37,6 +40,7 @@ The format is based on Keep a Changelog, and this project adheres to Semantic Ve
 - Fixed mesh size by appending `gParams.Dhi = gossipSubDhi`
 - Fix skipping partial withdrawals count.
 - recover from panics when writing the event stream [pr](https://github.com/prysmaticlabs/prysm/pull/14545)
+- `/eth/v1/node/syncing` returns `el_offline` false while the el is syncing.
 
 ### Security
 

beacon-chain/blockchain/process_block.go

@@ -404,6 +404,10 @@ func (s *Service) savePostStateInfo(ctx context.Context, r [32]byte, b interface
 		return errors.Wrapf(err, "could not save block from slot %d", b.Block().Slot())
 	}
 	if err := s.cfg.StateGen.SaveState(ctx, r, st); err != nil {
+		log.Warnf("Rolling back insertion of block with root %#x", r)
+		if err := s.cfg.BeaconDB.DeleteBlock(ctx, r); err != nil {
+			log.WithError(err).Errorf("Could not delete block with block root %#x", r)
+		}
 		return errors.Wrap(err, "could not save state")
 	}
 	return nil

beacon-chain/blockchain/receive_block.go

@@ -349,6 +349,9 @@ func (s *Service) ReceiveBlockBatch(ctx context.Context, blocks []blocks.ROBlock
 
 // HasBlock returns true if the block of the input root exists in initial sync blocks cache or DB.
 func (s *Service) HasBlock(ctx context.Context, root [32]byte) bool {
+	if s.BlockBeingSynced(root) {
+		return false
+	}
 	return s.hasBlockInInitSyncOrDB(ctx, root)
 }
 

beacon-chain/blockchain/receive_block_test.go

@@ -278,6 +278,8 @@ func TestService_HasBlock(t *testing.T) {
 	r, err = b.Block.HashTreeRoot()
 	require.NoError(t, err)
 	require.Equal(t, true, s.HasBlock(context.Background(), r))
+	s.blockBeingSynced.set(r)
+	require.Equal(t, false, s.HasBlock(context.Background(), r))
 }
 
 func TestCheckSaveHotStateDB_Enabling(t *testing.T) {

beacon-chain/execution/service.go

@@ -34,6 +34,7 @@ import (
 	contracts "github.com/prysmaticlabs/prysm/v5/contracts/deposit"
 	"github.com/prysmaticlabs/prysm/v5/encoding/bytesutil"
 	"github.com/prysmaticlabs/prysm/v5/monitoring/clientstats"
+	"github.com/prysmaticlabs/prysm/v5/monitoring/tracing/trace"
 	"github.com/prysmaticlabs/prysm/v5/network"
 	ethpb "github.com/prysmaticlabs/prysm/v5/proto/prysm/v1alpha1"
 	prysmTime "github.com/prysmaticlabs/prysm/v5/time"
@@ -41,6 +42,11 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
+const (
+	// EthSyncing request string for JSON-RPC.
+	EthSyncing = "eth_syncing"
+)
+
 var (
 	validDepositsCount = promauto.NewCounter(prometheus.CounterOpts{
 		Name: "powchain_valid_deposits_received",
@@ -79,6 +85,7 @@ type ChainInfoFetcher interface {
 	ExecutionClientConnected() bool
 	ExecutionClientEndpoint() string
 	ExecutionClientConnectionErr() error
+	IsExecutionClientSyncing(ctx context.Context) (bool, error)
 }
 
 // POWBlockFetcher defines a struct that can retrieve mainchain blocks.
@@ -303,6 +310,20 @@ func (s *Service) updateConnectedETH1(state bool) {
 	s.updateBeaconNodeStats()
 }
 
+// IsExecutionClientSyncing returns true if the execution client is syncing, otherwise false.
+
+func (s *Service) IsExecutionClientSyncing(ctx context.Context) (bool, error) {
+	ctx, span := trace.StartSpan(ctx, "powchain.engine-api-client.IsExecutionClientSyncing")
+	defer span.End()
+
+	var result bool
+	err := s.rpcClient.CallContext(ctx, &result, EthSyncing)
+	if err != nil {
+		return result, handleRPCError(err)
+	}
+	return result, nil
+}
+
 // refers to the latest eth1 block which follows the condition: eth1_timestamp +
 // SECONDS_PER_ETH1_BLOCK * ETH1_FOLLOW_DISTANCE <= current_unix_time
 func (s *Service) followedBlockHeight(ctx context.Context) (uint64, error) {

beacon-chain/execution/testing/mock_execution_chain.go

@@ -131,6 +131,10 @@ func (m *Chain) ExecutionClientConnectionErr() error {
 	return m.CurrError
 }
 
+func (m *Chain) IsExecutionClientSyncing(_ context.Context) (bool, error) {
+	return false, nil
+}
+
 func (m *Chain) ETH1Endpoints() []string {
 	return m.Endpoints
 }

beacon-chain/forkchoice/doubly-linked-tree/forkchoice.go

@@ -141,7 +141,14 @@ func (f *ForkChoice) InsertNode(ctx context.Context, state state.BeaconState, ro
 	}
 
 	jc, fc = f.store.pullTips(state, node, jc, fc)
-	return f.updateCheckpoints(ctx, jc, fc)
+	if err := f.updateCheckpoints(ctx, jc, fc); err != nil {
+		_, remErr := f.store.removeNode(ctx, node)
+		if remErr != nil {
+			log.WithError(remErr).Error("could not remove node")
+		}
+		return errors.Wrap(err, "could not update checkpoints")
+	}
+	return nil
 }
 
 // updateCheckpoints update the checkpoints when inserting a new node.

beacon-chain/forkchoice/doubly-linked-tree/forkchoice_test.go

@@ -3,6 +3,7 @@ package doublylinkedtree
 import (
 	"context"
 	"encoding/binary"
+	"errors"
 	"testing"
 	"time"
 
@@ -887,3 +888,16 @@ func TestForkchoiceParentRoot(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, zeroHash, root)
 }
+
+func TestForkChoice_CleanupInserting(t *testing.T) {
+	f := setup(0, 0)
+	ctx := context.Background()
+	st, blkRoot, err := prepareForkchoiceState(ctx, 1, indexToHash(1), params.BeaconConfig().ZeroHash, params.BeaconConfig().ZeroHash, 2, 2)
+	f.SetBalancesByRooter(func(_ context.Context, _ [32]byte) ([]uint64, error) {
+		return f.justifiedBalances, errors.New("mock err")
+	})
+
+	require.NoError(t, err)
+	require.NotNil(t, f.InsertNode(ctx, st, blkRoot))
+	require.Equal(t, false, f.HasNode(blkRoot))
+}

beacon-chain/forkchoice/doubly-linked-tree/store.go

@@ -107,7 +107,9 @@ func (s *Store) insert(ctx context.Context,
 			s.headNode = n
 			s.highestReceivedNode = n
 		} else {
-			return n, errInvalidParentRoot
+			delete(s.nodeByRoot, root)
+			delete(s.nodeByPayload, payloadHash)
+			return nil, errInvalidParentRoot
 		}
 	} else {
 		parent.children = append(parent.children, n)
@@ -128,7 +130,11 @@ func (s *Store) insert(ctx context.Context,
 		jEpoch := s.justifiedCheckpoint.Epoch
 		fEpoch := s.finalizedCheckpoint.Epoch
 		if err := s.treeRootNode.updateBestDescendant(ctx, jEpoch, fEpoch, slots.ToEpoch(currentSlot)); err != nil {
-			return n, err
+			_, remErr := s.removeNode(ctx, n)
+			if remErr != nil {
+				log.WithError(remErr).Error("could not remove node")
+			}
+			return nil, errors.Wrap(err, "could not update best descendants")
 		}
 	}
 	// Update metrics.

beacon-chain/forkchoice/doubly-linked-tree/store_test.go

@@ -525,3 +525,12 @@ func TestStore_TargetRootForEpoch(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, root4, target)
 }
+
+func TestStore_CleanupInserting(t *testing.T) {
+	f := setup(0, 0)
+	ctx := context.Background()
+	st, blkRoot, err := prepareForkchoiceState(ctx, 1, indexToHash(1), indexToHash(2), params.BeaconConfig().ZeroHash, 0, 0)
+	require.NoError(t, err)
+	require.NotNil(t, f.InsertNode(ctx, st, blkRoot))
+	require.Equal(t, false, f.HasNode(blkRoot))
+}

beacon-chain/rpc/eth/node/BUILD.bazel

@@ -28,6 +28,7 @@ go_library(
         "@com_github_ethereum_go_ethereum//common/hexutil:go_default_library",
         "@com_github_libp2p_go_libp2p//core/peer:go_default_library",
         "@com_github_pkg_errors//:go_default_library",
+        "@com_github_sirupsen_logrus//:go_default_library",
         "@org_golang_google_grpc//:go_default_library",
     ],
 )

beacon-chain/rpc/eth/node/handlers.go

@@ -14,6 +14,7 @@ import (
 	"github.com/prysmaticlabs/prysm/v5/network/httputil"
 	ethpb "github.com/prysmaticlabs/prysm/v5/proto/eth/v1"
 	"github.com/prysmaticlabs/prysm/v5/runtime/version"
+	log "github.com/sirupsen/logrus"
 )
 
 var (
@@ -37,6 +38,18 @@ func (s *Server) GetSyncStatus(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	isElOnline := s.ExecutionChainInfoFetcher.ExecutionClientConnected()
+	if !isElOnline {
+		isSyncing, err := s.ExecutionChainInfoFetcher.IsExecutionClientSyncing(ctx)
+		if err != nil {
+			log.WithField("error", err.Error()).Debug("failed to get execution client sync status")
+		} else if isSyncing {
+			isElOnline = true
+		} else {
+			log.Debug("execution client is not syncing and not connected")
+		}
+	}
+
 	headSlot := s.HeadFetcher.HeadSlot()
 	response := &structs.SyncStatusResponse{
 		Data: &structs.SyncStatusResponseData{
@@ -44,7 +57,7 @@ func (s *Server) GetSyncStatus(w http.ResponseWriter, r *http.Request) {
 			SyncDistance: strconv.FormatUint(uint64(s.GenesisTimeFetcher.CurrentSlot()-headSlot), 10),
 			IsSyncing:    s.SyncChecker.Syncing(),
 			IsOptimistic: isOptimistic,
-			ElOffline:    !s.ExecutionChainInfoFetcher.ExecutionClientConnected(),
+			ElOffline:    !isElOnline,
 		},
 	}
 	httputil.WriteJson(w, response)

beacon-chain/rpc/eth/node/handlers_test.go

@@ -56,7 +56,6 @@ func TestSyncStatus(t *testing.T) {
 
 	request := httptest.NewRequest(http.MethodGet, "http://example.com", nil)
 	writer := httptest.NewRecorder()
-	writer.Body = &bytes.Buffer{}
 
 	s.GetSyncStatus(writer, request)
 	assert.Equal(t, http.StatusOK, writer.Code)
@@ -67,6 +66,26 @@ func TestSyncStatus(t *testing.T) {
 	assert.Equal(t, "10", resp.Data.SyncDistance)
 	assert.Equal(t, true, resp.Data.IsSyncing)
 	assert.Equal(t, true, resp.Data.IsOptimistic)
+	assert.Equal(t, true, resp.Data.ElOffline)
+
+	// if execution client is syncing return offline false
+	s.ExecutionChainInfoFetcher = &testutil.MockExecutionChainInfoFetcher{Syncing: true}
+	writer = httptest.NewRecorder()
+	s.GetSyncStatus(writer, request)
+	assert.Equal(t, http.StatusOK, writer.Code)
+	resp = &structs.SyncStatusResponse{}
+	require.NoError(t, json.Unmarshal(writer.Body.Bytes(), resp))
+	require.NotNil(t, resp)
+	assert.Equal(t, false, resp.Data.ElOffline)
+
+	// if execution client is connected offline should be false
+	s.ExecutionChainInfoFetcher = &testutil.MockExecutionChainInfoFetcher{Connected: true}
+	writer = httptest.NewRecorder()
+	s.GetSyncStatus(writer, request)
+	assert.Equal(t, http.StatusOK, writer.Code)
+	resp = &structs.SyncStatusResponse{}
+	require.NoError(t, json.Unmarshal(writer.Body.Bytes(), resp))
+	require.NotNil(t, resp)
 	assert.Equal(t, false, resp.Data.ElOffline)
 }
 

beacon-chain/rpc/testutil/mock_exec_chain_info_fetcher.go

@@ -1,6 +1,7 @@
 package testutil
 
 import (
+	"context"
 	"math/big"
 )
 
@@ -8,14 +9,16 @@ import (
 type MockExecutionChainInfoFetcher struct {
 	CurrEndpoint string
 	CurrError    error
+	Syncing      bool
+	Connected    bool
 }
 
 func (*MockExecutionChainInfoFetcher) GenesisExecutionChainInfo() (uint64, *big.Int) {
 	return uint64(0), &big.Int{}
 }
 
-func (*MockExecutionChainInfoFetcher) ExecutionClientConnected() bool {
-	return true
+func (m *MockExecutionChainInfoFetcher) ExecutionClientConnected() bool {
+	return m.Connected
 }
 
 func (m *MockExecutionChainInfoFetcher) ExecutionClientEndpoint() string {
@@ -25,3 +28,7 @@ func (m *MockExecutionChainInfoFetcher) ExecutionClientEndpoint() string {
 func (m *MockExecutionChainInfoFetcher) ExecutionClientConnectionErr() error {
 	return m.CurrError
 }
+
+func (m *MockExecutionChainInfoFetcher) IsExecutionClientSyncing(_ context.Context) (bool, error) {
+	return m.Syncing, nil
+}