Insert Finalized Deposits In Another Routine (#8405)

* chk

* terence's review

* add test

Co-authored-by: Raul Jordan <raul@prysmaticlabs.com>

beacon-chain/blockchain/process_block.go

@@ -22,6 +22,9 @@ import (
 // A custom slot deadline for processing state slots in our cache.
 const slotDeadline = 5 * time.Second
 
+// A custom deadline for deposit trie insertion.
+const depositDeadline = 20 * time.Second
+
 // This defines size of the upper bound for initial sync block cache.
 var initialSyncBlockCacheSize = 2 * params.BeaconConfig().SlotsPerEpoch
 
@@ -161,23 +164,16 @@ func (s *Service) onBlock(ctx context.Context, signed *ethpb.SignedBeaconBlock,
 				return errors.Wrap(err, "could not save new justified")
 			}
 		}
-
-		// Update deposit cache.
-		finalizedState, err := s.stateGen.StateByRoot(ctx, fRoot)
-		if err != nil {
-			return errors.Wrap(err, "could not fetch finalized state")
-		}
-		// We update the cache up to the last deposit index in the finalized block's state.
-		// We can be confident that these deposits will be included in some block
-		// because the Eth1 follow distance makes such long-range reorgs extremely unlikely.
-		eth1DepositIndex := int64(finalizedState.Eth1Data().DepositCount - 1)
-		s.depositCache.InsertFinalizedDeposits(ctx, eth1DepositIndex)
-		if featureconfig.Get().EnablePruningDepositProofs {
-			// Deposit proofs are only used during state transition and can be safely removed to save space.
-			if err = s.depositCache.PruneProofs(ctx, eth1DepositIndex); err != nil {
-				return errors.Wrap(err, "could not prune deposit proofs")
+		go func() {
+			// Use a custom deadline here, since this method runs asynchronously.
+			// We ignore the parent method's context and instead create a new one
+			// with a custom deadline, therefore using the background context instead.
+			depCtx, cancel := context.WithTimeout(context.Background(), depositDeadline)
+			defer cancel()
+			if err := s.insertFinalizedDeposits(depCtx, fRoot); err != nil {
+				log.WithError(err).Error("Could not insert finalized deposits.")
 			}
-		}
+		}()
 	}
 
 	defer reportAttestationInclusion(b)

beacon-chain/blockchain/process_block_helpers.go

@@ -406,6 +406,30 @@ func (s *Service) fillInForkChoiceMissingBlocks(ctx context.Context, blk *ethpb.
 	return nil
 }
 
+// inserts finalized deposits into our finalized deposit trie.
+func (s *Service) insertFinalizedDeposits(ctx context.Context, fRoot [32]byte) error {
+	ctx, span := trace.StartSpan(ctx, "blockChain.insertFinalizedDeposits")
+	defer span.End()
+
+	// Update deposit cache.
+	finalizedState, err := s.stateGen.StateByRoot(ctx, fRoot)
+	if err != nil {
+		return errors.Wrap(err, "could not fetch finalized state")
+	}
+	// We update the cache up to the last deposit index in the finalized block's state.
+	// We can be confident that these deposits will be included in some block
+	// because the Eth1 follow distance makes such long-range reorgs extremely unlikely.
+	eth1DepositIndex := int64(finalizedState.Eth1Data().DepositCount - 1)
+	s.depositCache.InsertFinalizedDeposits(ctx, eth1DepositIndex)
+	if featureconfig.Get().EnablePruningDepositProofs {
+		// Deposit proofs are only used during state transition and can be safely removed to save space.
+		if err = s.depositCache.PruneProofs(ctx, eth1DepositIndex); err != nil {
+			return errors.Wrap(err, "could not prune deposit proofs")
+		}
+	}
+	return nil
+}
+
 // The deletes input attestations from the attestation pool, so proposers don't include them in a block for the future.
 func (s *Service) deletePoolAtts(atts []*ethpb.Attestation) error {
 	for _, att := range atts {

beacon-chain/blockchain/process_block_test.go

@@ -3,6 +3,8 @@ package blockchain
 import (
 	"context"
 	"fmt"
+	"math/big"
+	"strconv"
 	"testing"
 	"time"
 
@@ -20,6 +22,7 @@ import (
 	pb "github.com/prysmaticlabs/prysm/proto/beacon/p2p/v1"
 	"github.com/prysmaticlabs/prysm/shared/attestationutil"
 	"github.com/prysmaticlabs/prysm/shared/bytesutil"
+	"github.com/prysmaticlabs/prysm/shared/featureconfig"
 	"github.com/prysmaticlabs/prysm/shared/params"
 	"github.com/prysmaticlabs/prysm/shared/testutil"
 	"github.com/prysmaticlabs/prysm/shared/testutil/assert"
@@ -948,3 +951,51 @@ func TestOnBlock_CanFinalize(t *testing.T) {
 	require.Equal(t, types.Epoch(3), service.CurrentJustifiedCheckpt().Epoch)
 	require.Equal(t, types.Epoch(2), service.FinalizedCheckpt().Epoch)
 }
+
+func TestInsertFinalizedDeposits(t *testing.T) {
+	flags := featureconfig.Get()
+	flags.EnablePruningDepositProofs = true
+	reset := featureconfig.InitWithReset(flags)
+	defer reset()
+
+	ctx := context.Background()
+	beaconDB := testDB.SetupDB(t)
+	depositCache, err := depositcache.New()
+	require.NoError(t, err)
+	cfg := &Config{
+		BeaconDB:        beaconDB,
+		StateGen:        stategen.New(beaconDB),
+		ForkChoiceStore: protoarray.New(0, 0, [32]byte{}),
+		DepositCache:    depositCache,
+	}
+	service, err := New(ctx, cfg)
+	require.NoError(t, err)
+
+	gs, _ := testutil.DeterministicGenesisState(t, 32)
+	require.NoError(t, service.saveGenesisData(ctx, gs))
+	gBlk, err := service.beaconDB.GenesisBlock(ctx)
+	require.NoError(t, err)
+	gRoot, err := gBlk.Block.HashTreeRoot()
+	require.NoError(t, err)
+	service.finalizedCheckpt = &ethpb.Checkpoint{Root: gRoot[:]}
+	gs = gs.Copy()
+	assert.NoError(t, gs.SetEth1Data(&ethpb.Eth1Data{DepositCount: 10}))
+	assert.NoError(t, service.stateGen.SaveState(ctx, [32]byte{'m', 'o', 'c', 'k'}, gs))
+	zeroSig := [96]byte{}
+	for i := uint64(0); i < 4*params.BeaconConfig().SlotsPerEpoch; i++ {
+		root := []byte(strconv.Itoa(int(i)))
+		depositCache.InsertDeposit(ctx, &ethpb.Deposit{Data: &ethpb.Deposit_Data{
+			PublicKey:             bytesutil.FromBytes48([48]byte{}),
+			WithdrawalCredentials: params.BeaconConfig().ZeroHash[:],
+			Amount:                0,
+			Signature:             zeroSig[:],
+		}, Proof: [][]byte{root}}, 100+i, int64(i), bytesutil.ToBytes32(root))
+	}
+	assert.NoError(t, service.insertFinalizedDeposits(ctx, [32]byte{'m', 'o', 'c', 'k'}))
+	fDeposits := depositCache.FinalizedDeposits(ctx)
+	assert.Equal(t, 9, int(fDeposits.MerkleTrieIndex), "Finalized deposits not inserted correctly")
+	deps := depositCache.AllDeposits(ctx, big.NewInt(109))
+	for _, d := range deps {
+		assert.DeepEqual(t, [][]byte(nil), d.Proof, "Proofs are not empty")
+	}
+}