Disable Deposit Log processing after Deposit Requests are Activated (#15274)

* Disable Log Processing

* Changelog

* Kasey's review

---------

Co-authored-by: james-prysm <90280386+james-prysm@users.noreply.github.com>

beacon-chain/execution/log_processing_test.go

@@ -479,6 +479,93 @@ func TestProcessETH2GenesisLog_CorrectNumOfDeposits(t *testing.T) {
 	hook.Reset()
 }
 
+func TestProcessLogs_DepositRequestsStarted(t *testing.T) {
+	params.SetupTestConfigCleanup(t)
+	hook := logTest.NewGlobal()
+	testAcc, err := mock.Setup()
+	require.NoError(t, err, "Unable to set up simulated backend")
+	kvStore := testDB.SetupDB(t)
+	depositCache, err := depositsnapshot.New()
+	require.NoError(t, err)
+	server, endpoint, err := mockExecution.SetupRPCServer()
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		server.Stop()
+	})
+
+	web3Service, err := NewService(context.Background(),
+		WithHttpEndpoint(endpoint),
+		WithDepositContractAddress(testAcc.ContractAddr),
+		WithDatabase(kvStore),
+		WithDepositCache(depositCache),
+	)
+	require.NoError(t, err, "unable to setup web3 ETH1.0 chain service")
+	web3Service = setDefaultMocks(web3Service)
+	web3Service.depositContractCaller, err = contracts.NewDepositContractCaller(testAcc.ContractAddr, testAcc.Backend.Client())
+	require.NoError(t, err)
+	web3Service.rpcClient = &mockExecution.RPCClient{Backend: testAcc.Backend}
+	web3Service.httpLogger = testAcc.Backend.Client()
+	web3Service.latestEth1Data.LastRequestedBlock = 0
+	block, err := testAcc.Backend.Client().BlockByNumber(context.Background(), nil)
+	require.NoError(t, err)
+	web3Service.latestEth1Data.BlockHeight = block.NumberU64()
+	web3Service.latestEth1Data.BlockTime = block.Time()
+	bConfig := params.MinimalSpecConfig().Copy()
+	bConfig.MinGenesisTime = 0
+	bConfig.SecondsPerETH1Block = 1
+	params.OverrideBeaconConfig(bConfig)
+	nConfig := params.BeaconNetworkConfig()
+	nConfig.ContractDeploymentBlock = 0
+	params.OverrideBeaconNetworkConfig(nConfig)
+
+	testAcc.Backend.Commit()
+
+	totalNumOfDeposits := depositsReqForChainStart + 30
+
+	deposits, _, err := util.DeterministicDepositsAndKeys(uint64(totalNumOfDeposits))
+	require.NoError(t, err)
+	_, depositRoots, err := util.DeterministicDepositTrie(len(deposits))
+	require.NoError(t, err)
+	depositOffset := 5
+
+	// 64 Validators are used as size required for beacon-chain to start. This number
+	// is defined in the deposit contract as the number required for the testnet. The actual number
+	// is 2**14
+	for i := 0; i < totalNumOfDeposits; i++ {
+		data := deposits[i].Data
+		testAcc.TxOpts.Value = mock.Amount32Eth()
+		testAcc.TxOpts.GasLimit = 1000000
+		_, err = testAcc.Contract.Deposit(testAcc.TxOpts, data.PublicKey, data.WithdrawalCredentials, data.Signature, depositRoots[i])
+		require.NoError(t, err, "Could not deposit to deposit contract")
+		// pack 8 deposits into a block with an offset of
+		// 5
+		if (i+1)%8 == depositOffset {
+			testAcc.Backend.Commit()
+		}
+	}
+	// Forward the chain to account for the follow distance
+	for i := uint64(0); i < params.BeaconConfig().Eth1FollowDistance; i++ {
+		testAcc.Backend.Commit()
+	}
+	b, err := testAcc.Backend.Client().BlockByNumber(context.Background(), nil)
+	require.NoError(t, err)
+	web3Service.latestEth1Data.BlockHeight = b.NumberU64()
+	web3Service.latestEth1Data.BlockTime = b.Time()
+
+	// Set up our subscriber now to listen for the chain started event.
+	stateChannel := make(chan *feed.Event, 1)
+	stateSub := web3Service.cfg.stateNotifier.StateFeed().Subscribe(stateChannel)
+	defer stateSub.Unsubscribe()
+
+	web3Service.depositRequestsStarted = true
+	web3Service.initPOWService()
+	require.NoError(t, err)
+
+	require.Equal(t, int64(-1), web3Service.lastReceivedMerkleIndex, "Processed deposit logs even when requests are active")
+
+	hook.Reset()
+}
+
 func TestProcessETH2GenesisLog_LargePeriodOfNoLogs(t *testing.T) {
 	params.SetupTestConfigCleanup(t)
 	hook := logTest.NewGlobal()

beacon-chain/execution/service.go

@@ -16,6 +16,7 @@ import (
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/cache"
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/cache/depositsnapshot"
 	statefeed "github.com/OffchainLabs/prysm/v6/beacon-chain/core/feed/state"
+	"github.com/OffchainLabs/prysm/v6/beacon-chain/core/helpers"
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/core/transition"
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/db"
 	"github.com/OffchainLabs/prysm/v6/beacon-chain/execution/types"
@@ -141,6 +142,7 @@ type config struct {
 type Service struct {
 	connectedETH1           bool
 	isRunning               bool
+	depositRequestsStarted  bool
 	processingLock          sync.RWMutex
 	latestEth1DataLock      sync.RWMutex
 	cfg                     *config
@@ -205,7 +207,7 @@ func NewService(ctx context.Context, opts ...Option) (*Service, error) {
 			return nil, err
 		}
 	}
-
+	s.initDepositRequests()
 	eth1Data, err := s.validPowchainData(ctx)
 	if err != nil {
 		return nil, errors.Wrap(err, "unable to validate powchain data")
@@ -463,7 +465,9 @@ func safelyHandlePanic() {
 func (s *Service) handleETH1FollowDistance() {
 	defer safelyHandlePanic()
 	ctx := s.ctx
-
+	if s.depositRequestsStarted {
+		return
+	}
 	// use a 5 minutes timeout for block time, because the max mining time is 278 sec (block 7208027)
 	// (analyzed the time of the block from 2018-09-01 to 2019-02-13)
 	fiveMinutesTimeout := prysmTime.Now().Add(-5 * time.Minute)
@@ -530,25 +534,27 @@ func (s *Service) initPOWService() {
 			s.latestEth1Data.BlockTime = header.Time
 			s.latestEth1DataLock.Unlock()
 
-			if err := s.processPastLogs(ctx); err != nil {
-				err = errors.Wrap(err, "processPastLogs")
-				s.retryExecutionClientConnection(ctx, err)
-				errorLogger(
-					err,
-					"Unable to process past deposit contract logs, perhaps your execution client is not fully synced",
-				)
-				continue
-			}
-			// Cache eth1 headers from our voting period.
-			if err := s.cacheHeadersForEth1DataVote(ctx); err != nil {
-				err = errors.Wrap(err, "cacheHeadersForEth1DataVote")
-				s.retryExecutionClientConnection(ctx, err)
-				if errors.Is(err, errBlockTimeTooLate) {
-					log.WithError(err).Debug("Unable to cache headers for execution client votes")
-				} else {
-					errorLogger(err, "Unable to cache headers for execution client votes")
+			if !s.depositRequestsStarted {
+				if err := s.processPastLogs(ctx); err != nil {
+					err = errors.Wrap(err, "processPastLogs")
+					s.retryExecutionClientConnection(ctx, err)
+					errorLogger(
+						err,
+						"Unable to process past deposit contract logs, perhaps your execution client is not fully synced",
+					)
+					continue
+				}
+				// Cache eth1 headers from our voting period.
+				if err := s.cacheHeadersForEth1DataVote(ctx); err != nil {
+					err = errors.Wrap(err, "cacheHeadersForEth1DataVote")
+					s.retryExecutionClientConnection(ctx, err)
+					if errors.Is(err, errBlockTimeTooLate) {
+						log.WithError(err).Debug("Unable to cache headers for execution client votes")
+					} else {
+						errorLogger(err, "Unable to cache headers for execution client votes")
+					}
+					continue
 				}
-				continue
 			}
 			// Handle edge case with embedded genesis state by fetching genesis header to determine
 			// its height.
@@ -824,7 +830,7 @@ func (s *Service) validPowchainData(ctx context.Context) (*ethpb.ETH1ChainData,
 	if genState == nil || genState.IsNil() {
 		return eth1Data, nil
 	}
-	if eth1Data == nil || !eth1Data.ChainstartData.Chainstarted || !validateDepositContainers(eth1Data.DepositContainers) {
+	if s.depositRequestsStarted || eth1Data == nil || !eth1Data.ChainstartData.Chainstarted || !validateDepositContainers(eth1Data.DepositContainers) {
 		pbState, err := native.ProtobufBeaconStatePhase0(s.preGenesisState.ToProtoUnsafe())
 		if err != nil {
 			return nil, err
@@ -900,6 +906,15 @@ func (s *Service) removeStartupState() {
 	s.cfg.finalizedStateAtStartup = nil
 }
 
+func (s *Service) initDepositRequests() {
+	fState := s.cfg.finalizedStateAtStartup
+	isNil := fState == nil || fState.IsNil()
+	if isNil {
+		return
+	}
+	s.depositRequestsStarted = helpers.DepositRequestsStarted(fState)
+}
+
 func newBlobVerifierFromInitializer(ini *verification.Initializer) verification.NewBlobVerifier {
 	return func(b blocks.ROBlob, reqs []verification.Requirement) verification.BlobVerifier {
 		return ini.NewBlobVerifier(b, reqs)