Compare commits

...

21 Commits

Author SHA1 Message Date
prestonvanloon
2838683245 Revert "Revert "revert costing change""
This reverts commit 64db5ad361.

I'm done looking at this.
2022-09-02 13:05:20 -05:00
Preston Van Loon
476d188fcc Update beacon-chain/blockchain/error.go
Co-authored-by: Potuz <potuz@prysmaticlabs.com>
2022-09-02 13:04:14 -05:00
prestonvanloon
64db5ad361 Revert "revert costing change"
This reverts commit d0c7cedbba.
2022-09-02 13:03:52 -05:00
prestonvanloon
d0c7cedbba revert costing change 2022-09-02 13:02:45 -05:00
prestonvanloon
b20937e97a go mod tidy 2022-09-02 09:44:45 -05:00
prestonvanloon
52a581362c restore godoc comment 2022-09-02 09:25:19 -05:00
prestonvanloon
537142f79e ensure db doesn't have junk root 2022-09-02 09:23:51 -05:00
prestonvanloon
abb32e5f27 Skip saving bad state summaries, return multiple errors 2022-09-02 09:22:32 -05:00
prestonvanloon
08c3d149bf Tests for pending blocks cache insertion 2022-09-02 09:01:55 -05:00
prestonvanloon
0d5b7d093a Add test requiring a block to exist 2022-09-02 08:53:34 -05:00
prestonvanloon
80f30a0d5b more PR feedback 2022-09-02 08:38:49 -05:00
prestonvanloon
9580f64fea Minimal changes. Terence feedback 2022-09-02 08:34:00 -05:00
nisdas
b3c3f7f40e hack 2022-09-02 17:39:22 +08:00
nisdas
49f2e12030 fix tests 2022-09-02 17:19:56 +08:00
prestonvanloon
9aa4407daf A few more passing test cases 2022-09-01 22:44:24 -05:00
prestonvanloon
05b3ef1fcd Fix mosts tests 2022-09-01 22:18:57 -05:00
Preston Van Loon
8d086e433d Merge branch 'develop' into ss-missing-blocks 2022-09-01 15:53:35 -05:00
Preston Van Loon
3833f2770e Merge branch 'develop' into ss-missing-blocks 2022-09-01 12:53:33 -05:00
prestonvanloon
e6c04e5dad fix interfacechecker 2022-09-01 12:22:07 -05:00
prestonvanloon
5e05398daf Lots of changes... 2022-09-01 11:51:17 -05:00
prestonvanloon
299fa3494a Enforce block exists in db before saving state summary 2022-09-01 09:35:18 -05:00
32 changed files with 404 additions and 127 deletions

View File

@@ -530,13 +530,13 @@ func TestService_IsOptimisticForRoot_DB_ProtoArray(t *testing.T) {
optimisticBlock := util.NewBeaconBlock()
optimisticBlock.Block.Slot = 97
optimisticRoot, err := optimisticBlock.Block.HashTreeRoot()
optimisticRoot, err := util.SaveBlock(t, ctx, beaconDB, optimisticBlock).Block().HashTreeRoot()
require.NoError(t, err)
util.SaveBlock(t, context.Background(), beaconDB, optimisticBlock)
validatedBlock := util.NewBeaconBlock()
validatedBlock.Block.Slot = 9
validatedRoot, err := validatedBlock.Block.HashTreeRoot()
validatedRoot, err := util.SaveBlock(t, ctx, beaconDB, validatedBlock).Block().HashTreeRoot()
require.NoError(t, err)
util.SaveBlock(t, context.Background(), beaconDB, validatedBlock)
@@ -563,10 +563,11 @@ func TestService_IsOptimisticForRoot_DB_ProtoArray(t *testing.T) {
require.NoError(t, err)
require.Equal(t, false, validated)
// Before the first finalized epoch, finalized root could be zeros.
validatedCheckpoint = &ethpb.Checkpoint{Root: params.BeaconConfig().ZeroHash[:]}
r, err := util.SaveBlock(t, ctx, beaconDB, util.NewBeaconBlock()).Block().HashTreeRoot()
require.NoError(t, err)
validatedCheckpoint = &ethpb.Checkpoint{Root: r[:]}
require.NoError(t, beaconDB.SaveGenesisBlockRoot(ctx, br))
require.NoError(t, beaconDB.SaveStateSummary(context.Background(), &ethpb.StateSummary{Root: params.BeaconConfig().ZeroHash[:], Slot: 10}))
require.NoError(t, beaconDB.SaveStateSummary(context.Background(), &ethpb.StateSummary{Root: r[:], Slot: 10}))
require.NoError(t, beaconDB.SaveLastValidatedCheckpoint(ctx, validatedCheckpoint))
require.NoError(t, beaconDB.SaveStateSummary(context.Background(), &ethpb.StateSummary{Root: optimisticRoot[:], Slot: 11}))
@@ -621,10 +622,11 @@ func TestService_IsOptimisticForRoot_DB_DoublyLinkedTree(t *testing.T) {
require.NoError(t, err)
require.Equal(t, false, validated)
// Before the first finalized epoch, finalized root could be zeros.
validatedCheckpoint = &ethpb.Checkpoint{Root: params.BeaconConfig().ZeroHash[:]}
r, err := util.SaveBlock(t, ctx, beaconDB, util.NewBeaconBlock()).Block().HashTreeRoot()
require.NoError(t, err)
validatedCheckpoint = &ethpb.Checkpoint{Root: r[:]}
require.NoError(t, beaconDB.SaveGenesisBlockRoot(ctx, br))
require.NoError(t, beaconDB.SaveStateSummary(context.Background(), &ethpb.StateSummary{Root: params.BeaconConfig().ZeroHash[:], Slot: 10}))
require.NoError(t, beaconDB.SaveStateSummary(context.Background(), &ethpb.StateSummary{Root: r[:], Slot: 10}))
require.NoError(t, beaconDB.SaveLastValidatedCheckpoint(ctx, validatedCheckpoint))
require.NoError(t, beaconDB.SaveStateSummary(context.Background(), &ethpb.StateSummary{Root: optimisticRoot[:], Slot: 11}))

View File

@@ -23,6 +23,8 @@ var (
errNotOptimisticCandidate = errors.New("block is not suitable for optimistic sync")
// errBlockNotFoundInCacheOrDB is returned when a block is not found in the cache or DB.
errBlockNotFoundInCacheOrDB = errors.New("block not found in cache or db")
// errNilBlockInCache is returned when a nil block is returned from the cache.
ErrNilBlockInCache = errors.New("nil block returned from the cache")
// errNilStateFromStategen is returned when a nil state is returned from the state generator.
errNilStateFromStategen = errors.New("justified state can't be nil")
// errWSBlockNotFound is returned when a block is not found in the WS cache or DB.

View File

@@ -1083,14 +1083,14 @@ func Test_UpdateLastValidatedCheckpoint(t *testing.T) {
require.NoError(t, fcs.InsertNode(ctx, state, blkRoot))
fcs.SetOriginRoot(genesisRoot)
genesisSummary := &ethpb.StateSummary{
Root: genesisStateRoot[:],
Root: genesisRoot[:],
Slot: 0,
}
require.NoError(t, beaconDB.SaveStateSummary(ctx, genesisSummary))
// Get last validated checkpoint
origCheckpoint, err := service.cfg.BeaconDB.LastValidatedCheckpoint(ctx)
require.NoError(t, err)
// Set last validated checkpoint to a junk root to prevent issues
// with saving it due to the state summary.
origCheckpoint := &ethpb.Checkpoint{Root: genesisRoot[:], Epoch: 0}
require.NoError(t, beaconDB.SaveLastValidatedCheckpoint(ctx, origCheckpoint))
// Optimistic finalized checkpoint

View File

@@ -15,6 +15,7 @@ import (
"github.com/prysmaticlabs/prysm/v3/config/features"
"github.com/prysmaticlabs/prysm/v3/config/params"
"github.com/prysmaticlabs/prysm/v3/consensus-types/blocks"
consensusblocks "github.com/prysmaticlabs/prysm/v3/consensus-types/blocks"
types "github.com/prysmaticlabs/prysm/v3/consensus-types/primitives"
"github.com/prysmaticlabs/prysm/v3/encoding/bytesutil"
ethpbv1 "github.com/prysmaticlabs/prysm/v3/proto/eth/v1"
@@ -160,10 +161,16 @@ func TestCacheJustifiedStateBalances_CanCache(t *testing.T) {
ctx := context.Background()
state, _ := util.DeterministicGenesisState(t, 100)
r := [32]byte{'a'}
require.NoError(t, service.cfg.BeaconDB.SaveStateSummary(context.Background(), &ethpb.StateSummary{Root: r[:]}))
require.NoError(t, service.cfg.BeaconDB.SaveState(context.Background(), state, r))
balances, err := service.justifiedBalances.get(ctx, r)
newBlock := util.NewBeaconBlock()
newBlock.Block.Slot = 20
rt, err := newBlock.Block.HashTreeRoot()
assert.NoError(t, err)
wrappedBlk, err := consensusblocks.NewSignedBeaconBlock(newBlock)
assert.NoError(t, err)
assert.NoError(t, service.cfg.BeaconDB.SaveBlock(ctx, wrappedBlk))
require.NoError(t, service.cfg.BeaconDB.SaveStateSummary(context.Background(), &ethpb.StateSummary{Root: rt[:]}))
require.NoError(t, service.cfg.BeaconDB.SaveState(context.Background(), state, rt))
balances, err := service.justifiedBalances.get(ctx, rt)
require.NoError(t, err)
require.DeepEqual(t, balances, state.Balances(), "Incorrect justified balances")
}

View File

@@ -75,6 +75,17 @@ func (s *Service) getInitSyncBlocks() []interfaces.SignedBeaconBlock {
return blks
}
// getInitSyncBlockFromCache returns a block from the initial sync blocks cache. This method
func (s *Service) getInitSyncBlockFromCache(r [32]byte) (interfaces.SignedBeaconBlock, error) {
s.initSyncBlocksLock.RLock()
defer s.initSyncBlocksLock.RUnlock()
b := s.initSyncBlocks[r]
if b.IsNil() {
return nil, ErrNilBlockInCache
}
return b, nil
}
// This clears out the initial sync blocks cache.
func (s *Service) clearInitSyncBlocks() {
s.initSyncBlocksLock.Lock()

View File

@@ -307,7 +307,11 @@ func TestStore_SaveCheckpointState(t *testing.T) {
s, err := util.NewBeaconState()
require.NoError(t, err)
err = s.SetFinalizedCheckpoint(&ethpb.Checkpoint{Root: bytesutil.PadTo([]byte{'A'}, fieldparams.RootLength)})
b0 := util.NewBeaconBlock()
b0.Block.Slot = 0
r0, err := util.SaveBlock(t, ctx, beaconDB, b0).Block().HashTreeRoot()
require.NoError(t, err)
err = s.SetFinalizedCheckpoint(&ethpb.Checkpoint{Root: r0[:]})
require.NoError(t, err)
val := &ethpb.Validator{
PublicKey: bytesutil.PadTo([]byte("foo"), 48),
@@ -317,20 +321,27 @@ func TestStore_SaveCheckpointState(t *testing.T) {
require.NoError(t, err)
err = s.SetBalances([]uint64{0})
require.NoError(t, err)
r := [32]byte{'g'}
require.NoError(t, service.cfg.BeaconDB.SaveState(ctx, s, r))
require.NoError(t, service.cfg.BeaconDB.SaveState(ctx, s, r0))
cp1 := &ethpb.Checkpoint{Epoch: 1, Root: bytesutil.PadTo([]byte{'A'}, fieldparams.RootLength)}
require.NoError(t, service.cfg.BeaconDB.SaveState(ctx, s, bytesutil.ToBytes32([]byte{'A'})))
require.NoError(t, service.cfg.BeaconDB.SaveStateSummary(ctx, &ethpb.StateSummary{Root: bytesutil.PadTo([]byte{'A'}, fieldparams.RootLength)}))
b1 := util.NewBeaconBlock()
b1.Block.Slot = 1
r1, err := util.SaveBlock(t, ctx, beaconDB, b1).Block().HashTreeRoot()
require.NoError(t, err)
cp1 := &ethpb.Checkpoint{Epoch: 1, Root: r1[:]}
require.NoError(t, service.cfg.BeaconDB.SaveState(ctx, s, r1))
require.NoError(t, service.cfg.BeaconDB.SaveStateSummary(ctx, &ethpb.StateSummary{Root: r1[:]}))
s1, err := service.getAttPreState(ctx, cp1)
require.NoError(t, err)
assert.Equal(t, 1*params.BeaconConfig().SlotsPerEpoch, s1.Slot(), "Unexpected state slot")
cp2 := &ethpb.Checkpoint{Epoch: 2, Root: bytesutil.PadTo([]byte{'B'}, fieldparams.RootLength)}
require.NoError(t, service.cfg.BeaconDB.SaveState(ctx, s, bytesutil.ToBytes32([]byte{'B'})))
require.NoError(t, service.cfg.BeaconDB.SaveStateSummary(ctx, &ethpb.StateSummary{Root: bytesutil.PadTo([]byte{'B'}, fieldparams.RootLength)}))
b2 := util.NewBeaconBlock()
b2.Block.Slot = 2
r2, err := util.SaveBlock(t, ctx, beaconDB, b2).Block().HashTreeRoot()
require.NoError(t, err)
cp2 := &ethpb.Checkpoint{Epoch: 2, Root: r2[:]}
require.NoError(t, service.cfg.BeaconDB.SaveState(ctx, s, r2))
require.NoError(t, service.cfg.BeaconDB.SaveStateSummary(ctx, &ethpb.StateSummary{Root: r2[:]}))
s2, err := service.getAttPreState(ctx, cp2)
require.NoError(t, err)
assert.Equal(t, 2*params.BeaconConfig().SlotsPerEpoch, s2.Slot(), "Unexpected state slot")
@@ -348,9 +359,13 @@ func TestStore_SaveCheckpointState(t *testing.T) {
assert.Equal(t, 2*params.BeaconConfig().SlotsPerEpoch, s2.Slot(), "Unexpected state slot")
require.NoError(t, s.SetSlot(params.BeaconConfig().SlotsPerEpoch+1))
cp3 := &ethpb.Checkpoint{Epoch: 1, Root: bytesutil.PadTo([]byte{'C'}, fieldparams.RootLength)}
require.NoError(t, service.cfg.BeaconDB.SaveState(ctx, s, bytesutil.ToBytes32([]byte{'C'})))
require.NoError(t, service.cfg.BeaconDB.SaveStateSummary(ctx, &ethpb.StateSummary{Root: bytesutil.PadTo([]byte{'C'}, fieldparams.RootLength)}))
b3 := util.NewBeaconBlock()
b3.Block.Slot = 3
r3, err := util.SaveBlock(t, ctx, beaconDB, b3).Block().HashTreeRoot()
require.NoError(t, err)
cp3 := &ethpb.Checkpoint{Epoch: 1, Root: r3[:]}
require.NoError(t, service.cfg.BeaconDB.SaveState(ctx, s, r3))
require.NoError(t, service.cfg.BeaconDB.SaveStateSummary(ctx, &ethpb.StateSummary{Root: r3[:]}))
s3, err := service.getAttPreState(ctx, cp3)
require.NoError(t, err)
assert.Equal(t, s.Slot(), s3.Slot(), "Unexpected state slot")

View File

@@ -408,13 +408,15 @@ func (s *Service) onBlockBatch(ctx context.Context, blks []interfaces.SignedBeac
tracing.AnnotateError(span, err)
return err
}
if err := s.cfg.BeaconDB.SaveStateSummary(ctx, &ethpb.StateSummary{
if err := s.cfg.BeaconDB.SaveStateSummariesWithPendingBlocks(ctx, []*ethpb.StateSummary{{
Slot: b.Block().Slot(),
Root: blockRoots[i][:],
}); err != nil {
}}, s.getInitSyncBlockFromCache); err != nil {
tracing.AnnotateError(span, err)
return err
}
if i > 0 && jCheckpoints[i].Epoch > jCheckpoints[i-1].Epoch {
if err := s.cfg.BeaconDB.SaveJustifiedCheckpoint(ctx, jCheckpoints[i]); err != nil {
tracing.AnnotateError(span, err)
@@ -445,6 +447,17 @@ func (s *Service) onBlockBatch(ctx context.Context, blks []interfaces.SignedBeac
}
for r, st := range boundaries {
// Ensure the cached block is saved to db before saving state boundary.
if !s.cfg.BeaconDB.HasBlock(ctx, r) {
b, err := s.getInitSyncBlockFromCache(r)
if err != nil || b.IsNil() {
log.WithField("block root", fmt.Sprintf("%#x", bytesutil.Trunc(r[:]))).Warn("Could not find block for boundary state root in cache")
} else {
if err := s.cfg.BeaconDB.SaveBlock(ctx, b); err != nil {
return err
}
}
}
if err := s.cfg.StateGen.SaveState(ctx, r, st); err != nil {
return err
}

View File

@@ -391,6 +391,7 @@ func TestStore_OnBlockBatch_NotifyNewPayload(t *testing.T) {
}
err = service.onBlockBatch(ctx, blks, blkRoots)
require.NoError(t, err)
require.NoError(t, service.Stop())
}
func TestCachedPreState_CanGetFromStateSummary_ProtoArray(t *testing.T) {
@@ -1350,7 +1351,14 @@ func TestInsertFinalizedDeposits(t *testing.T) {
gs = gs.Copy()
assert.NoError(t, gs.SetEth1Data(&ethpb.Eth1Data{DepositCount: 10}))
assert.NoError(t, gs.SetEth1DepositIndex(8))
assert.NoError(t, service.cfg.StateGen.SaveState(ctx, [32]byte{'m', 'o', 'c', 'k'}, gs))
newBlock := util.NewBeaconBlock()
newBlock.Block.Slot = 20
rt, err := newBlock.Block.HashTreeRoot()
assert.NoError(t, err)
wrappedBlk, err := consensusblocks.NewSignedBeaconBlock(newBlock)
assert.NoError(t, err)
assert.NoError(t, service.cfg.BeaconDB.SaveBlock(ctx, wrappedBlk))
assert.NoError(t, service.cfg.StateGen.SaveState(ctx, rt, gs))
zeroSig := [96]byte{}
for i := uint64(0); i < uint64(4*params.BeaconConfig().SlotsPerEpoch); i++ {
root := []byte(strconv.Itoa(int(i)))
@@ -1361,7 +1369,7 @@ func TestInsertFinalizedDeposits(t *testing.T) {
Signature: zeroSig[:],
}, Proof: [][]byte{root}}, 100+i, int64(i), bytesutil.ToBytes32(root)))
}
assert.NoError(t, service.insertFinalizedDeposits(ctx, [32]byte{'m', 'o', 'c', 'k'}))
assert.NoError(t, service.insertFinalizedDeposits(ctx, rt))
fDeposits := depositCache.FinalizedDeposits(ctx)
assert.Equal(t, 7, int(fDeposits.MerkleTrieIndex), "Finalized deposits not inserted correctly")
deps := depositCache.AllDeposits(ctx, big.NewInt(107))
@@ -1384,11 +1392,27 @@ func TestInsertFinalizedDeposits_MultipleFinalizedRoutines(t *testing.T) {
gs = gs.Copy()
assert.NoError(t, gs.SetEth1Data(&ethpb.Eth1Data{DepositCount: 7}))
assert.NoError(t, gs.SetEth1DepositIndex(6))
assert.NoError(t, service.cfg.StateGen.SaveState(ctx, [32]byte{'m', 'o', 'c', 'k'}, gs))
newBlock := util.NewBeaconBlock()
newBlock.Block.Slot = 20
rt, err := newBlock.Block.HashTreeRoot()
assert.NoError(t, err)
wrappedBlk, err := consensusblocks.NewSignedBeaconBlock(newBlock)
assert.NoError(t, err)
assert.NoError(t, service.cfg.BeaconDB.SaveBlock(ctx, wrappedBlk))
assert.NoError(t, service.cfg.StateGen.SaveState(ctx, rt, gs))
gs2 := gs.Copy()
assert.NoError(t, gs2.SetEth1Data(&ethpb.Eth1Data{DepositCount: 15}))
assert.NoError(t, gs2.SetEth1DepositIndex(13))
assert.NoError(t, service.cfg.StateGen.SaveState(ctx, [32]byte{'m', 'o', 'c', 'k', '2'}, gs2))
newBlock2 := util.NewBeaconBlock()
newBlock2.Block.Slot = 30
rt2, err := newBlock2.Block.HashTreeRoot()
assert.NoError(t, err)
wrappedBlk2, err := consensusblocks.NewSignedBeaconBlock(newBlock2)
assert.NoError(t, err)
assert.NoError(t, service.cfg.BeaconDB.SaveBlock(ctx, wrappedBlk2))
assert.NoError(t, service.cfg.StateGen.SaveState(ctx, rt2, gs2))
zeroSig := [96]byte{}
for i := uint64(0); i < uint64(4*params.BeaconConfig().SlotsPerEpoch); i++ {
root := []byte(strconv.Itoa(int(i)))
@@ -1402,7 +1426,7 @@ func TestInsertFinalizedDeposits_MultipleFinalizedRoutines(t *testing.T) {
// Insert 3 deposits before hand.
depositCache.InsertFinalizedDeposits(ctx, 2)
assert.NoError(t, service.insertFinalizedDeposits(ctx, [32]byte{'m', 'o', 'c', 'k'}))
assert.NoError(t, service.insertFinalizedDeposits(ctx, rt))
fDeposits := depositCache.FinalizedDeposits(ctx)
assert.Equal(t, 5, int(fDeposits.MerkleTrieIndex), "Finalized deposits not inserted correctly")
@@ -1412,7 +1436,7 @@ func TestInsertFinalizedDeposits_MultipleFinalizedRoutines(t *testing.T) {
}
// Insert New Finalized State with higher deposit count.
assert.NoError(t, service.insertFinalizedDeposits(ctx, [32]byte{'m', 'o', 'c', 'k', '2'}))
assert.NoError(t, service.insertFinalizedDeposits(ctx, rt2))
fDeposits = depositCache.FinalizedDeposits(ctx)
assert.Equal(t, 12, int(fDeposits.MerkleTrieIndex), "Finalized deposits not inserted correctly")
deps = depositCache.AllDeposits(ctx, big.NewInt(112))
@@ -1425,12 +1449,12 @@ func TestRemoveBlockAttestationsInPool_Canonical(t *testing.T) {
genesis, keys := util.DeterministicGenesisState(t, 64)
b, err := util.GenerateFullBlock(genesis, keys, util.DefaultBlockGenConfig(), 1)
assert.NoError(t, err)
r, err := b.Block.HashTreeRoot()
require.NoError(t, err)
ctx := context.Background()
beaconDB := testDB.SetupDB(t)
service := setupBeaconChain(t, beaconDB)
r, err := util.SaveBlock(t, ctx, beaconDB, b).Block().HashTreeRoot()
require.NoError(t, err)
require.NoError(t, service.cfg.BeaconDB.SaveStateSummary(ctx, &ethpb.StateSummary{Root: r[:]}))
require.NoError(t, service.cfg.BeaconDB.SaveGenesisBlockRoot(ctx, r))

View File

@@ -119,8 +119,11 @@ func setupBeaconChain(t *testing.T, beaconDB db.Database) *Service {
require.NoError(t, err)
stateGen := stategen.New(beaconDB)
// Safe a state in stategen to purposes of testing a service stop / shutdown.
require.NoError(t, stateGen.SaveState(ctx, bytesutil.ToBytes32(bState.FinalizedCheckpoint().Root), bState))
// Save a state and block in stategen to purposes of testing a service stop / shutdown.
r, err := util.SaveBlock(t, ctx, beaconDB, util.NewBeaconBlock()).Block().HashTreeRoot()
require.NoError(t, err)
require.NoError(t, stateGen.SaveState(ctx, r, bState))
opts := []Option{
WithDatabase(beaconDB),
@@ -192,8 +195,6 @@ func TestChainStartStop_GenesisZeroHashes(t *testing.T) {
require.NoError(t, beaconDB.SaveState(ctx, s, blkRoot))
require.NoError(t, beaconDB.SaveGenesisBlockRoot(ctx, blkRoot))
require.NoError(t, beaconDB.SaveBlock(ctx, wsb))
require.NoError(t, beaconDB.SaveJustifiedCheckpoint(ctx, &ethpb.Checkpoint{Root: params.BeaconConfig().ZeroHash[:]}))
require.NoError(t, beaconDB.SaveFinalizedCheckpoint(ctx, &ethpb.Checkpoint{Root: blkRoot[:]}))
chainService.cfg.FinalizedStateAtStartUp = s
// Test the start function.
chainService.Start()
@@ -381,7 +382,7 @@ func TestChainService_SaveHeadNoDB(t *testing.T) {
}
blk := util.NewBeaconBlock()
blk.Block.Slot = 1
r, err := blk.HashTreeRoot()
r, err := util.SaveBlock(t, ctx, beaconDB, blk).Block().HashTreeRoot()
require.NoError(t, err)
newState, err := util.NewBeaconState()
require.NoError(t, err)

View File

@@ -75,6 +75,7 @@ type NoHeadAccessDatabase interface {
DeleteStates(ctx context.Context, blockRoots [][32]byte) error
SaveStateSummary(ctx context.Context, summary *ethpb.StateSummary) error
SaveStateSummaries(ctx context.Context, summaries []*ethpb.StateSummary) error
SaveStateSummariesWithPendingBlocks(ctx context.Context, summaries []*ethpb.StateSummary, blockCache func([32]byte) (interfaces.SignedBeaconBlock, error)) error
// Checkpoint operations.
SaveJustifiedCheckpoint(ctx context.Context, checkpoint *ethpb.Checkpoint) error
SaveFinalizedCheckpoint(ctx context.Context, checkpoint *ethpb.Checkpoint) error

View File

@@ -73,6 +73,7 @@ go_library(
"@io_etcd_go_bbolt//:go_default_library",
"@io_opencensus_go//trace:go_default_library",
"@org_golang_google_protobuf//proto:go_default_library",
"@org_uber_go_multierr//:go_default_library",
],
)

View File

@@ -296,32 +296,40 @@ func (s *Store) SaveBlocks(ctx context.Context, blks []interfaces.SignedBeaconBl
indicesByBucket := createBlockIndicesFromBlock(ctx, blk.Block())
indicesForBlocks[i] = indicesByBucket
}
return s.db.Update(func(tx *bolt.Tx) error {
if err := s.db.Update(func(tx *bolt.Tx) error {
bkt := tx.Bucket(blocksBucket)
for i, blk := range blks {
for i := range blks {
if existingBlock := bkt.Get(blockRoots[i]); existingBlock != nil {
continue
}
if err := updateValueForIndices(ctx, indicesForBlocks[i], blockRoots[i], tx); err != nil {
return errors.Wrap(err, "could not update DB indices")
}
if features.Get().EnableOnlyBlindedBeaconBlocks {
blindedBlock, err := blk.ToBlinded()
if err != nil {
if !errors.Is(err, blocks.ErrUnsupportedVersion) {
return err
}
} else {
blk = blindedBlock
}
}
s.blockCache.Set(string(blockRoots[i]), blk, int64(len(encodedBlocks[i])))
if err := bkt.Put(blockRoots[i], encodedBlocks[i]); err != nil {
return err
}
}
return nil
})
}); err != nil {
return err
}
// Populate cache only after successful db update.
for i, blk := range blks {
if features.Get().EnableOnlyBlindedBeaconBlocks {
blindedBlock, err := blk.ToBlinded()
if err != nil {
if !errors.Is(err, blocks.ErrUnsupportedVersion) {
return err
}
} else {
blk = blindedBlock
}
}
s.blockCache.Set(string(blockRoots[i]), blk, int64(len(encodedBlocks[i])))
}
return nil
}
// SaveHeadBlockRoot to the db.

View File

@@ -9,6 +9,7 @@ var ErrDeleteJustifiedAndFinalized = errors.New("cannot delete finalized block o
// indicate that a value couldn't be found.
var ErrNotFound = errors.New("not found in db")
var ErrNotFoundState = errors.Wrap(ErrNotFound, "state not found")
var ErrNotFoundBlock = errors.Wrap(ErrNotFound, "block not found")
// ErrNotFoundOriginBlockRoot is an error specifically for the origin block root getter
var ErrNotFoundOriginBlockRoot = errors.Wrap(ErrNotFound, "OriginBlockRoot")

View File

@@ -2,11 +2,15 @@ package kv
import (
"context"
"fmt"
"github.com/pkg/errors"
"github.com/prysmaticlabs/prysm/v3/consensus-types/interfaces"
"github.com/prysmaticlabs/prysm/v3/encoding/bytesutil"
ethpb "github.com/prysmaticlabs/prysm/v3/proto/prysm/v1alpha1"
bolt "go.etcd.io/bbolt"
"go.opencensus.io/trace"
"go.uber.org/multierr"
)
// SaveStateSummary saves a state summary object to the DB.
@@ -21,10 +25,20 @@ func (s *Store) SaveStateSummary(ctx context.Context, summary *ethpb.StateSummar
func (s *Store) SaveStateSummaries(ctx context.Context, summaries []*ethpb.StateSummary) error {
ctx, span := trace.StartSpan(ctx, "BeaconDB.SaveStateSummaries")
defer span.End()
return s.SaveStateSummariesWithPendingBlocks(ctx, summaries, nil)
}
// SaveStateSummariesWithPendingBlocks saves state summary objects to the DB.
func (s *Store) SaveStateSummariesWithPendingBlocks(ctx context.Context, summaries []*ethpb.StateSummary, blockCache func([32]byte) (interfaces.SignedBeaconBlock, error)) error {
ctx, span := trace.StartSpan(ctx, "BeaconDB.SaveStateSummariesWithPendingBlocks")
defer span.End()
// When we reach the state summary cache prune count,
// dump the cached state summaries to the DB.
if s.stateSummaryCache.len() >= stateSummaryCachePruneCount {
if err := s.ensureBlocksSaved(ctx, blockCache); err != nil {
return err
}
if err := s.saveCachedStateSummariesDB(ctx); err != nil {
return err
}
@@ -37,6 +51,25 @@ func (s *Store) SaveStateSummaries(ctx context.Context, summaries []*ethpb.State
return nil
}
func (s *Store) ensureBlocksSaved(ctx context.Context, blockCache func([32]byte) (interfaces.SignedBeaconBlock, error)) error {
if blockCache == nil {
return nil
}
summaries := s.stateSummaryCache.getAll()
blocks := make([]interfaces.SignedBeaconBlock, 0, len(summaries))
for _, ss := range summaries {
if s.HasBlock(ctx, bytesutil.ToBytes32(ss.Root)) {
log.WithField("blockRoot", fmt.Sprintf("%#x", ss.Root)).Trace("Block already saved")
continue
}
if b, err := blockCache(bytesutil.ToBytes32(ss.Root)); err == nil && !b.IsNil() {
blocks = append(blocks, b)
}
}
log.WithField("blocks", len(blocks)).WithField("summaries", len(summaries)).Debug("Saving blocks from state summary cache")
return s.SaveBlocks(ctx, blocks)
}
// StateSummary returns the state summary object from the db using input block root.
func (s *Store) StateSummary(ctx context.Context, blockRoot [32]byte) (*ethpb.StateSummary, error) {
ctx, span := trace.StartSpan(ctx, "BeaconDB.StateSummary")
@@ -96,10 +129,16 @@ func (s *Store) saveCachedStateSummariesDB(ctx context.Context) error {
}
encs[i] = enc
}
var errs []error
if err := s.db.Update(func(tx *bolt.Tx) error {
bucket := tx.Bucket(stateSummaryBucket)
bBkt := tx.Bucket(blocksBucket)
ssBkt := tx.Bucket(stateSummaryBucket)
for i, s := range summaries {
if err := bucket.Put(s.Root, encs[i]); err != nil {
if bBkt.Get(s.Root) == nil {
errs = append(errs, errors.Wrapf(ErrNotFoundBlock, "failed to save state summary with block root %#x", s.Root))
continue
}
if err := ssBkt.Put(s.Root, encs[i]); err != nil {
return err
}
}
@@ -108,7 +147,7 @@ func (s *Store) saveCachedStateSummariesDB(ctx context.Context) error {
return err
}
s.stateSummaryCache.clear()
return nil
return multierr.Combine(errs...)
}
// deleteStateSummary deletes a state summary object from the db using input block root.

View File

@@ -4,18 +4,27 @@ import (
"context"
"testing"
"github.com/prysmaticlabs/prysm/v3/consensus-types/blocks"
"github.com/prysmaticlabs/prysm/v3/consensus-types/interfaces"
types "github.com/prysmaticlabs/prysm/v3/consensus-types/primitives"
"github.com/prysmaticlabs/prysm/v3/encoding/bytesutil"
ethpb "github.com/prysmaticlabs/prysm/v3/proto/prysm/v1alpha1"
"github.com/prysmaticlabs/prysm/v3/testing/assert"
"github.com/prysmaticlabs/prysm/v3/testing/require"
"github.com/prysmaticlabs/prysm/v3/testing/util"
)
func TestStateSummary_CanSaveRetrieve(t *testing.T) {
db := setupDB(t)
ctx := context.Background()
r1 := bytesutil.ToBytes32([]byte{'A'})
r2 := bytesutil.ToBytes32([]byte{'B'})
b1 := util.NewBeaconBlock()
b1.Block.Slot = 1
r1, err := util.SaveBlock(t, ctx, db, b1).Block().HashTreeRoot()
require.NoError(t, err)
b2 := util.NewBeaconBlock()
b2.Block.Slot = 2
r2, err := util.SaveBlock(t, ctx, db, b2).Block().HashTreeRoot()
require.NoError(t, err)
s1 := &ethpb.StateSummary{Slot: 1, Root: r1[:]}
// State summary should not exist yet.
@@ -43,23 +52,130 @@ func TestStateSummary_CanSaveRetrieve(t *testing.T) {
func TestStateSummary_CacheToDB(t *testing.T) {
db := setupDB(t)
ctx := context.Background()
summaries := make([]*ethpb.StateSummary, stateSummaryCachePruneCount-1)
roots := make([][32]byte, stateSummaryCachePruneCount-1)
for i := range summaries {
summaries[i] = &ethpb.StateSummary{Slot: types.Slot(i), Root: bytesutil.PadTo(bytesutil.Uint64ToBytesLittleEndian(uint64(i)), 32)}
b := util.NewBeaconBlock()
b.Block.Slot = types.Slot(i)
b.Block.Body.Graffiti = bytesutil.PadTo([]byte{byte(i)}, 32)
r, err := util.SaveBlock(t, ctx, db, b).Block().HashTreeRoot()
require.NoError(t, err)
summaries[i] = &ethpb.StateSummary{Slot: types.Slot(i), Root: r[:]}
roots[i] = r
}
require.NoError(t, db.SaveStateSummaries(context.Background(), summaries))
require.Equal(t, db.stateSummaryCache.len(), stateSummaryCachePruneCount-1)
require.NoError(t, db.SaveStateSummary(context.Background(), &ethpb.StateSummary{Slot: 1000, Root: []byte{'a', 'b'}}))
b := util.NewBeaconBlock()
b.Block.Slot = types.Slot(1000)
r, err := util.SaveBlock(t, ctx, db, b).Block().HashTreeRoot()
require.NoError(t, err)
require.NoError(t, db.SaveStateSummary(context.Background(), &ethpb.StateSummary{Slot: 1000, Root: r[:]}))
require.Equal(t, db.stateSummaryCache.len(), stateSummaryCachePruneCount)
require.NoError(t, db.SaveStateSummary(context.Background(), &ethpb.StateSummary{Slot: 1001, Root: []byte{'c', 'd'}}))
b = util.NewBeaconBlock()
b.Block.Slot = 1001
r, err = util.SaveBlock(t, ctx, db, b).Block().HashTreeRoot()
require.NoError(t, err)
require.NoError(t, db.SaveStateSummary(context.Background(), &ethpb.StateSummary{Slot: 1001, Root: r[:]}))
require.Equal(t, db.stateSummaryCache.len(), 1)
for _, r := range roots {
require.Equal(t, true, db.HasStateSummary(context.Background(), r))
}
}
func TestStateSummary_CacheToDB_FailsIfMissingBlock(t *testing.T) {
db := setupDB(t)
ctx := context.Background()
summaries := make([]*ethpb.StateSummary, stateSummaryCachePruneCount-1)
for i := range summaries {
r := bytesutil.Uint64ToBytesLittleEndian(uint64(i))
require.Equal(t, true, db.HasStateSummary(context.Background(), bytesutil.ToBytes32(r)))
b := util.NewBeaconBlock()
b.Block.Slot = types.Slot(i)
b.Block.Body.Graffiti = bytesutil.PadTo([]byte{byte(i)}, 32)
r, err := util.SaveBlock(t, ctx, db, b).Block().HashTreeRoot()
require.NoError(t, err)
summaries[i] = &ethpb.StateSummary{Slot: types.Slot(i), Root: r[:]}
}
require.NoError(t, db.SaveStateSummaries(context.Background(), summaries))
require.Equal(t, db.stateSummaryCache.len(), stateSummaryCachePruneCount-1)
junkRoot := [32]byte{1, 2, 3}
require.NoError(t, db.SaveStateSummary(context.Background(), &ethpb.StateSummary{Slot: 1000, Root: junkRoot[:]}))
require.Equal(t, db.stateSummaryCache.len(), stateSummaryCachePruneCount)
// Next insertion causes the buffer to flush.
b := util.NewBeaconBlock()
b.Block.Slot = 1001
r, err := util.SaveBlock(t, ctx, db, b).Block().HashTreeRoot()
require.NoError(t, err)
require.ErrorIs(t, db.SaveStateSummary(context.Background(), &ethpb.StateSummary{Slot: 1001, Root: r[:]}), ErrNotFoundBlock)
require.Equal(t, db.HasStateSummary(ctx, junkRoot), false)
require.NoError(t, db.deleteStateSummary(junkRoot)) // Delete bad summary or db will throw an error on test cleanup.
}
func TestStateSummary_CacheToDB_UsesProvidedBlockCache(t *testing.T) {
db := setupDB(t)
ctx := context.Background()
summaries := make([]*ethpb.StateSummary, stateSummaryCachePruneCount-1)
bCache := make(map[[32]byte]interfaces.SignedBeaconBlock, stateSummaryCachePruneCount-1)
for i := range summaries {
b := util.NewBeaconBlock()
b.Block.Slot = types.Slot(i)
b.Block.Body.Graffiti = bytesutil.PadTo([]byte{byte(i)}, 32)
wsb, err := blocks.NewSignedBeaconBlock(b)
require.NoError(t, err)
r, err := wsb.Block().HashTreeRoot()
require.NoError(t, err)
bCache[r] = wsb
summaries[i] = &ethpb.StateSummary{Slot: types.Slot(i), Root: r[:]}
}
// Simulated initial sync block cache.
pendingBlocksFn := func(r [32]byte) (interfaces.SignedBeaconBlock, error) {
b, ok := bCache[r]
if !ok {
return nil, ErrNotFoundBlock
}
return b, nil
}
require.NoError(t, db.SaveStateSummariesWithPendingBlocks(context.Background(), summaries, pendingBlocksFn))
require.Equal(t, db.stateSummaryCache.len(), stateSummaryCachePruneCount-1)
b := util.NewBeaconBlock()
b.Block.Slot = types.Slot(1000)
r, err := util.SaveBlock(t, ctx, db, b).Block().HashTreeRoot()
require.NoError(t, err)
require.NoError(t, db.SaveStateSummariesWithPendingBlocks(context.Background(), []*ethpb.StateSummary{{Slot: 1000, Root: r[:]}}, pendingBlocksFn))
require.Equal(t, db.stateSummaryCache.len(), stateSummaryCachePruneCount)
// Next insertion causes the buffer to flush.
b = util.NewBeaconBlock()
b.Block.Slot = 1001
r, err = util.SaveBlock(t, ctx, db, b).Block().HashTreeRoot()
require.NoError(t, err)
require.NoError(t, db.SaveStateSummariesWithPendingBlocks(context.Background(), []*ethpb.StateSummary{{Slot: 1001, Root: r[:]}}, pendingBlocksFn))
require.Equal(t, db.stateSummaryCache.len(), 1)
// All blocks in the bCache should have been saved.
for r := range bCache {
require.Equal(t, true, db.HasBlock(ctx, r), "Block %#x was not saved to db", r)
}
}

View File

@@ -473,6 +473,7 @@ func TestInitDepositCacheWithFinalization_OK(t *testing.T) {
emptyState, err := util.NewBeaconState()
require.NoError(t, err)
util.SaveBlock(t, context.Background(), s.cfg.beaconDB, headBlock)
require.NoError(t, s.cfg.beaconDB.SaveGenesisBlockRoot(context.Background(), headRoot))
require.NoError(t, s.cfg.beaconDB.SaveState(context.Background(), emptyState, headRoot))
require.NoError(t, stateGen.SaveState(context.Background(), headRoot, emptyState))

View File

@@ -53,6 +53,7 @@ go_test(
"//beacon-chain/core/altair:go_default_library",
"//beacon-chain/core/feed:go_default_library",
"//beacon-chain/core/feed/state:go_default_library",
"//beacon-chain/db/iface:go_default_library",
"//beacon-chain/db/testing:go_default_library",
"//beacon-chain/state/stategen:go_default_library",
"//config/params:go_default_library",

View File

@@ -3,6 +3,7 @@ package monitor
import (
"bytes"
"context"
"fmt"
"testing"
"github.com/prysmaticlabs/go-bitfield"
@@ -33,7 +34,7 @@ func TestGetAttestingIndices(t *testing.T) {
func TestProcessIncludedAttestationTwoTracked(t *testing.T) {
hook := logTest.NewGlobal()
s := setupService(t)
s, _ := setupService(t)
state, _ := util.DeterministicGenesisStateAltair(t, 256)
require.NoError(t, state.SetSlot(2))
require.NoError(t, state.SetCurrentParticipationBits(bytes.Repeat([]byte{0xff}, 13)))
@@ -66,7 +67,7 @@ func TestProcessUnaggregatedAttestationStateNotCached(t *testing.T) {
hook := logTest.NewGlobal()
ctx := context.Background()
s := setupService(t)
s, _ := setupService(t)
state, _ := util.DeterministicGenesisStateAltair(t, 256)
require.NoError(t, state.SetSlot(2))
header := state.LatestBlockHeader()
@@ -98,13 +99,13 @@ func TestProcessUnaggregatedAttestationStateCached(t *testing.T) {
ctx := context.Background()
hook := logTest.NewGlobal()
s := setupService(t)
s, db := setupService(t)
state, _ := util.DeterministicGenesisStateAltair(t, 256)
participation := []byte{0xff, 0xff, 0x01, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}
require.NoError(t, state.SetCurrentParticipationBits(participation))
root := [32]byte{}
copy(root[:], "hello-world")
root, err := util.SaveBlock(t, ctx, db, util.NewBeaconBlock()).Block().HashTreeRoot()
require.NoError(t, err)
att := &ethpb.Attestation{
Data: &ethpb.AttestationData{
@@ -124,8 +125,9 @@ func TestProcessUnaggregatedAttestationStateCached(t *testing.T) {
}
require.NoError(t, s.config.StateGen.SaveState(ctx, root, state))
s.processUnaggregatedAttestation(context.Background(), att)
wanted1 := "\"Processed unaggregated attestation\" Head=0x68656c6c6f2d Slot=1 Source=0x68656c6c6f2d Target=0x68656c6c6f2d ValidatorIndex=2 prefix=monitor"
wanted2 := "\"Processed unaggregated attestation\" Head=0x68656c6c6f2d Slot=1 Source=0x68656c6c6f2d Target=0x68656c6c6f2d ValidatorIndex=12 prefix=monitor"
rootStr := fmt.Sprintf("%#x", bytesutil.Trunc(root[:]))
wanted1 := fmt.Sprintf("\"Processed unaggregated attestation\" Head=%s Slot=1 Source=%s Target=%s ValidatorIndex=2 prefix=monitor", rootStr, rootStr, rootStr)
wanted2 := fmt.Sprintf("\"Processed unaggregated attestation\" Head=%s Slot=1 Source=%s Target=%s ValidatorIndex=12 prefix=monitor", rootStr, rootStr, rootStr)
require.LogsContain(t, hook, wanted1)
require.LogsContain(t, hook, wanted2)
}
@@ -135,7 +137,7 @@ func TestProcessAggregatedAttestationStateNotCached(t *testing.T) {
hook := logTest.NewGlobal()
ctx := context.Background()
s := setupService(t)
s, _ := setupService(t)
state, _ := util.DeterministicGenesisStateAltair(t, 256)
require.NoError(t, state.SetSlot(2))
header := state.LatestBlockHeader()
@@ -170,13 +172,13 @@ func TestProcessAggregatedAttestationStateNotCached(t *testing.T) {
func TestProcessAggregatedAttestationStateCached(t *testing.T) {
hook := logTest.NewGlobal()
ctx := context.Background()
s := setupService(t)
s, db := setupService(t)
state, _ := util.DeterministicGenesisStateAltair(t, 256)
participation := []byte{0xff, 0xff, 0x01, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}
require.NoError(t, state.SetCurrentParticipationBits(participation))
root := [32]byte{}
copy(root[:], "hello-world")
root, err := util.SaveBlock(t, ctx, db, util.NewBeaconBlock()).Block().HashTreeRoot()
require.NoError(t, err)
att := &ethpb.AggregateAttestationAndProof{
AggregatorIndex: 2,
@@ -200,14 +202,15 @@ func TestProcessAggregatedAttestationStateCached(t *testing.T) {
require.NoError(t, s.config.StateGen.SaveState(ctx, root, state))
s.processAggregatedAttestation(ctx, att)
require.LogsContain(t, hook, "\"Processed attestation aggregation\" AggregatorIndex=2 BeaconBlockRoot=0x68656c6c6f2d Slot=1 SourceRoot=0x68656c6c6f2d TargetRoot=0x68656c6c6f2d prefix=monitor")
require.LogsContain(t, hook, "\"Processed aggregated attestation\" Head=0x68656c6c6f2d Slot=1 Source=0x68656c6c6f2d Target=0x68656c6c6f2d ValidatorIndex=2 prefix=monitor")
require.LogsDoNotContain(t, hook, "\"Processed aggregated attestation\" Head=0x68656c6c6f2d Slot=1 Source=0x68656c6c6f2d Target=0x68656c6c6f2d ValidatorIndex=12 prefix=monitor")
rootStr := fmt.Sprintf("%#x", bytesutil.Trunc(root[:]))
require.LogsContain(t, hook, fmt.Sprintf("\"Processed attestation aggregation\" AggregatorIndex=2 BeaconBlockRoot=%s Slot=1 SourceRoot=%s TargetRoot=%s prefix=monitor", rootStr, rootStr, rootStr))
require.LogsContain(t, hook, fmt.Sprintf("\"Processed aggregated attestation\" Head=%s Slot=1 Source=%s Target=%s ValidatorIndex=2 prefix=monitor", rootStr, rootStr, rootStr))
require.LogsDoNotContain(t, hook, fmt.Sprintf("\"Processed aggregated attestation\" Head=%s Slot=1 Source=%s Target=%s ValidatorIndex=12 prefix=monitor", rootStr, rootStr, rootStr))
}
func TestProcessAttestations(t *testing.T) {
hook := logTest.NewGlobal()
s := setupService(t)
s, _ := setupService(t)
ctx := context.Background()
state, _ := util.DeterministicGenesisStateAltair(t, 256)
require.NoError(t, state.SetSlot(2))

View File

@@ -168,7 +168,7 @@ func TestProcessProposedBlock(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
hook := logTest.NewGlobal()
s := setupService(t)
s, _ := setupService(t)
beaconState, _ := util.DeterministicGenesisState(t, 256)
root := [32]byte{}
copy(root[:], "hello-world")
@@ -199,7 +199,7 @@ func TestProcessBlock_AllEventsTrackedVals(t *testing.T) {
genConfig.FullSyncAggregate = true
b, err := util.GenerateFullBlockAltair(genesis, keys, genConfig, 1)
require.NoError(t, err)
s := setupService(t)
s, db := setupService(t)
pubKeys := make([][]byte, 3)
pubKeys[0] = genesis.Validators()[0].PublicKey
@@ -223,7 +223,7 @@ func TestProcessBlock_AllEventsTrackedVals(t *testing.T) {
s.RUnlock()
s.updateSyncCommitteeTrackedVals(genesis)
root, err := b.GetBlock().HashTreeRoot()
root, err := util.SaveBlock(t, ctx, db, b).Block().HashTreeRoot()
require.NoError(t, err)
require.NoError(t, s.config.StateGen.SaveState(ctx, root, genesis))
wanted1 := fmt.Sprintf("\"Proposed beacon block was included\" BalanceChange=100000000 BlockRoot=%#x NewBalance=32000000000 ParentRoot=0xf732eaeb7fae ProposerIndex=15 Slot=1 Version=1 prefix=monitor", bytesutil.Trunc(root[:]))
@@ -241,7 +241,7 @@ func TestProcessBlock_AllEventsTrackedVals(t *testing.T) {
func TestLogAggregatedPerformance(t *testing.T) {
hook := logTest.NewGlobal()
s := setupService(t)
s, _ := setupService(t)
s.logAggregatedPerformance()
time.Sleep(3000 * time.Millisecond)

View File

@@ -13,7 +13,7 @@ import (
func TestProcessSyncCommitteeContribution(t *testing.T) {
hook := logTest.NewGlobal()
s := setupService(t)
s, _ := setupService(t)
contrib := &ethpb.SignedContributionAndProof{
Message: &ethpb.ContributionAndProof{
@@ -28,7 +28,7 @@ func TestProcessSyncCommitteeContribution(t *testing.T) {
func TestProcessSyncAggregate(t *testing.T) {
hook := logTest.NewGlobal()
s := setupService(t)
s, _ := setupService(t)
beaconState, _ := util.DeterministicGenesisStateAltair(t, 256)
block := &ethpb.BeaconBlockAltair{

View File

@@ -11,6 +11,7 @@ import (
"github.com/prysmaticlabs/prysm/v3/beacon-chain/core/altair"
"github.com/prysmaticlabs/prysm/v3/beacon-chain/core/feed"
statefeed "github.com/prysmaticlabs/prysm/v3/beacon-chain/core/feed/state"
"github.com/prysmaticlabs/prysm/v3/beacon-chain/db/iface"
testDB "github.com/prysmaticlabs/prysm/v3/beacon-chain/db/testing"
"github.com/prysmaticlabs/prysm/v3/beacon-chain/state/stategen"
"github.com/prysmaticlabs/prysm/v3/consensus-types/blocks"
@@ -22,7 +23,7 @@ import (
logTest "github.com/sirupsen/logrus/hooks/test"
)
func setupService(t *testing.T) *Service {
func setupService(t *testing.T) (*Service, iface.Database) {
beaconDB := testDB.SetupDB(t)
state, _ := util.DeterministicGenesisStateAltair(t, 256)
@@ -100,7 +101,7 @@ func setupService(t *testing.T) *Service {
aggregatedPerformance: aggregatedPerformance,
trackedSyncCommitteeIndices: trackedSyncCommitteeIndices,
lastSyncedEpoch: 0,
}
}, beaconDB
}
func TestTrackedIndex(t *testing.T) {
@@ -116,7 +117,7 @@ func TestTrackedIndex(t *testing.T) {
func TestUpdateSyncCommitteeTrackedVals(t *testing.T) {
hook := logTest.NewGlobal()
s := setupService(t)
s, _ := setupService(t)
state, _ := util.DeterministicGenesisStateAltair(t, 1024)
s.updateSyncCommitteeTrackedVals(state)
@@ -138,7 +139,7 @@ func TestNewService(t *testing.T) {
func TestStart(t *testing.T) {
hook := logTest.NewGlobal()
s := setupService(t)
s, _ := setupService(t)
stateChannel := make(chan *feed.Event, 1)
stateSub := s.config.StateNotifier.StateFeed().Subscribe(stateChannel)
defer stateSub.Unsubscribe()
@@ -180,7 +181,7 @@ func TestStart(t *testing.T) {
func TestInitializePerformanceStructures(t *testing.T) {
hook := logTest.NewGlobal()
ctx := context.Background()
s := setupService(t)
s, _ := setupService(t)
state, err := s.config.HeadFetcher.HeadState(ctx)
require.NoError(t, err)
epoch := slots.ToEpoch(state.Slot())
@@ -222,7 +223,7 @@ func TestInitializePerformanceStructures(t *testing.T) {
func TestMonitorRoutine(t *testing.T) {
ctx := context.Background()
hook := logTest.NewGlobal()
s := setupService(t)
s, db := setupService(t)
stateChannel := make(chan *feed.Event, 1)
stateSub := s.config.StateNotifier.StateFeed().Subscribe(stateChannel)
@@ -242,7 +243,7 @@ func TestMonitorRoutine(t *testing.T) {
genConfig := util.DefaultBlockGenConfig()
block, err := util.GenerateFullBlockAltair(genesis, keys, genConfig, 1)
require.NoError(t, err)
root, err := block.GetBlock().HashTreeRoot()
root, err := util.SaveBlock(t, ctx, db, block).Block().HashTreeRoot()
require.NoError(t, err)
require.NoError(t, s.config.StateGen.SaveState(ctx, root, genesis))
@@ -266,7 +267,7 @@ func TestMonitorRoutine(t *testing.T) {
}
func TestWaitForSync(t *testing.T) {
s := setupService(t)
s, _ := setupService(t)
stateChannel := make(chan *feed.Event, 1)
stateSub := s.config.StateNotifier.StateFeed().Subscribe(stateChannel)
defer stateSub.Unsubscribe()
@@ -290,7 +291,7 @@ func TestWaitForSync(t *testing.T) {
func TestRun(t *testing.T) {
hook := logTest.NewGlobal()
s := setupService(t)
s, _ := setupService(t)
stateChannel := make(chan *feed.Event, 1)
stateSub := s.config.StateNotifier.StateFeed().Subscribe(stateChannel)

View File

@@ -501,8 +501,12 @@ func TestServer_ListIndexedAttestations_GenesisEpoch(t *testing.T) {
db := dbTest.SetupDB(t)
helpers.ClearCache()
ctx := context.Background()
targetRoot1 := bytesutil.ToBytes32([]byte("root"))
targetRoot2 := bytesutil.ToBytes32([]byte("root2"))
targetRoot1, err := util.SaveBlock(t, ctx, db, util.NewBeaconBlock()).Block().HashTreeRoot()
require.NoError(t, err)
b := util.NewBeaconBlock()
b.Block.ParentRoot = targetRoot1[:]
targetRoot2, err := util.SaveBlock(t, ctx, db, b).Block().HashTreeRoot()
require.NoError(t, err)
count := params.BeaconConfig().SlotsPerEpoch
atts := make([]*ethpb.Attestation, 0, count)
@@ -571,7 +575,7 @@ func TestServer_ListIndexedAttestations_GenesisEpoch(t *testing.T) {
HeadFetcher: &chainMock.ChainService{State: state},
StateGen: stategen.New(db),
}
err := db.SaveStateSummary(ctx, &ethpb.StateSummary{
err = db.SaveStateSummary(ctx, &ethpb.StateSummary{
Root: targetRoot1[:],
Slot: 1,
})
@@ -610,7 +614,8 @@ func TestServer_ListIndexedAttestations_OldEpoch(t *testing.T) {
helpers.ClearCache()
ctx := context.Background()
blockRoot := bytesutil.ToBytes32([]byte("root"))
blockRoot, err := util.SaveBlock(t, ctx, db, util.NewBeaconBlock()).Block().HashTreeRoot()
require.NoError(t, err)
count := params.BeaconConfig().SlotsPerEpoch
atts := make([]*ethpb.Attestation, 0, count)
epoch := types.Epoch(50)

View File

@@ -1542,12 +1542,15 @@ func TestServer_GetValidatorParticipation_CurrentAndPrevEpoch(t *testing.T) {
b.Block.Slot = 16
util.SaveBlock(t, ctx, beaconDB, b)
bRoot, err := b.Block.HashTreeRoot()
require.NoError(t, err)
bRoot2, err := util.SaveBlock(t, ctx, beaconDB, b).Block().HashTreeRoot()
require.NoError(t, err)
require.NoError(t, beaconDB.SaveStateSummary(ctx, &ethpb.StateSummary{Root: bRoot[:]}))
require.NoError(t, beaconDB.SaveStateSummary(ctx, &ethpb.StateSummary{Root: params.BeaconConfig().ZeroHash[:]}))
require.NoError(t, beaconDB.SaveStateSummary(ctx, &ethpb.StateSummary{Root: bRoot2[:]}))
require.NoError(t, beaconDB.SaveGenesisBlockRoot(ctx, bRoot))
require.NoError(t, err)
require.NoError(t, beaconDB.SaveState(ctx, headState, bRoot))
require.NoError(t, beaconDB.SaveState(ctx, headState, params.BeaconConfig().ZeroHash))
require.NoError(t, beaconDB.SaveState(ctx, headState, bRoot2))
m := &mock.ChainService{State: headState}
offset := int64(params.BeaconConfig().SlotsPerEpoch.Mul(params.BeaconConfig().SecondsPerSlot))

View File

@@ -58,7 +58,9 @@ func TestServer_GetAttestationInclusionSlot(t *testing.T) {
}
s, _ := util.DeterministicGenesisState(t, 2048)
tr := [32]byte{'a'}
b := util.NewBeaconBlock()
tr, err := util.SaveBlock(t, ctx, bs.BeaconDB, b).Block().HashTreeRoot()
require.NoError(t, err)
require.NoError(t, bs.StateGen.SaveState(ctx, tr, s))
c, err := helpers.BeaconCommitteeFromState(context.Background(), s, 1, 0)
require.NoError(t, err)
@@ -73,7 +75,7 @@ func TestServer_GetAttestationInclusionSlot(t *testing.T) {
AggregationBits: bitfield.Bitlist{0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x01},
Signature: make([]byte, fieldparams.BLSSignatureLength),
}
b := util.NewBeaconBlock()
b = util.NewBeaconBlock()
b.Block.Slot = 2
b.Block.Body.Attestations = []*ethpb.Attestation{a}
util.SaveBlock(t, ctx, bs.BeaconDB, b)

View File

@@ -67,7 +67,8 @@ func Test_processQueuedBlocks_DetectsDoubleProposals(t *testing.T) {
blksQueue: newBlocksQueue(),
}
parentRoot := bytesutil.ToBytes32([]byte("parent"))
parentRoot, err := util.SaveBlock(t, ctx, beaconDB, util.NewBeaconBlock()).Block().HashTreeRoot()
require.NoError(t, err)
err = s.serviceCfg.StateGen.SaveState(ctx, parentRoot, beaconState)
require.NoError(t, err)

View File

@@ -11,7 +11,6 @@ import (
"github.com/prysmaticlabs/prysm/v3/beacon-chain/state/stategen"
"github.com/prysmaticlabs/prysm/v3/config/params"
"github.com/prysmaticlabs/prysm/v3/crypto/bls"
"github.com/prysmaticlabs/prysm/v3/encoding/bytesutil"
ethpb "github.com/prysmaticlabs/prysm/v3/proto/prysm/v1alpha1"
"github.com/prysmaticlabs/prysm/v3/testing/require"
"github.com/prysmaticlabs/prysm/v3/testing/util"
@@ -157,7 +156,8 @@ func TestService_processProposerSlashings(t *testing.T) {
},
}
parentRoot := bytesutil.ToBytes32([]byte("parent"))
parentRoot, err := util.SaveBlock(t, ctx, beaconDB, util.NewBeaconBlock()).Block().HashTreeRoot()
require.NoError(t, err)
err = s.serviceCfg.StateGen.SaveState(ctx, parentRoot, beaconState)
require.NoError(t, err)

View File

@@ -62,7 +62,9 @@ func TestStateByRootIfCachedNoCopy_HotState(t *testing.T) {
service := New(beaconDB)
beaconState, _ := util.DeterministicGenesisState(t, 32)
r := [32]byte{'A'}
b := util.NewBeaconBlock()
r, err := util.SaveBlock(t, ctx, beaconDB, b).Block().HashTreeRoot()
require.NoError(t, err)
require.NoError(t, service.beaconDB.SaveStateSummary(ctx, &ethpb.StateSummary{Root: r[:]}))
service.hotStateCache.put(r, beaconState)
@@ -102,6 +104,7 @@ func TestStateByRoot_HotStateUsingEpochBoundaryCacheNoReplay(t *testing.T) {
beaconState, _ := util.DeterministicGenesisState(t, 32)
require.NoError(t, beaconState.SetSlot(10))
blk := util.NewBeaconBlock()
util.SaveBlock(t, ctx, beaconDB, blk)
blkRoot, err := blk.Block.HashTreeRoot()
require.NoError(t, err)
require.NoError(t, service.beaconDB.SaveStateSummary(ctx, &ethpb.StateSummary{Root: blkRoot[:]}))
@@ -143,7 +146,8 @@ func TestStateByRoot_HotStateCached(t *testing.T) {
service := New(beaconDB)
beaconState, _ := util.DeterministicGenesisState(t, 32)
r := [32]byte{'A'}
r, err := util.SaveBlock(t, ctx, beaconDB, util.NewBeaconBlock()).Block().HashTreeRoot()
require.NoError(t, err)
require.NoError(t, service.beaconDB.SaveStateSummary(ctx, &ethpb.StateSummary{Root: r[:]}))
service.hotStateCache.put(r, beaconState)
@@ -223,7 +227,8 @@ func TestStateByRootInitialSync_UseCache(t *testing.T) {
service := New(beaconDB)
beaconState, _ := util.DeterministicGenesisState(t, 32)
r := [32]byte{'A'}
r, err := util.SaveBlock(t, ctx, beaconDB, util.NewBeaconBlock()).Block().HashTreeRoot()
require.NoError(t, err)
require.NoError(t, service.beaconDB.SaveStateSummary(ctx, &ethpb.StateSummary{Root: r[:]}))
service.hotStateCache.put(r, beaconState)

View File

@@ -74,13 +74,6 @@ func (s *State) saveStateByRoot(ctx context.Context, blockRoot [32]byte, st stat
return nil
}
// Only on an epoch boundary slot, save epoch boundary state in epoch boundary root state cache.
if slots.IsEpochStart(st.Slot()) {
if err := s.epochBoundaryStateCache.put(blockRoot, st); err != nil {
return err
}
}
// On an intermediate slot, save state summary.
if err := s.beaconDB.SaveStateSummary(ctx, &ethpb.StateSummary{
Slot: st.Slot(),
@@ -89,6 +82,13 @@ func (s *State) saveStateByRoot(ctx context.Context, blockRoot [32]byte, st stat
return err
}
// Only on an epoch boundary slot, save epoch boundary state in epoch boundary root state cache.
if slots.IsEpochStart(st.Slot()) {
if err := s.epochBoundaryStateCache.put(blockRoot, st); err != nil {
return err
}
}
// Store the copied state in the hot state cache.
s.hotStateCache.put(blockRoot, st)

View File

@@ -6,6 +6,7 @@ import (
testDB "github.com/prysmaticlabs/prysm/v3/beacon-chain/db/testing"
"github.com/prysmaticlabs/prysm/v3/config/params"
"github.com/prysmaticlabs/prysm/v3/encoding/bytesutil"
"github.com/prysmaticlabs/prysm/v3/testing/assert"
"github.com/prysmaticlabs/prysm/v3/testing/require"
"github.com/prysmaticlabs/prysm/v3/testing/util"
@@ -22,7 +23,8 @@ func TestSaveState_HotStateCanBeSaved(t *testing.T) {
// This goes to hot section, verify it can save on epoch boundary.
require.NoError(t, beaconState.SetSlot(params.BeaconConfig().SlotsPerEpoch))
r := [32]byte{'a'}
r, err := util.SaveBlock(t, ctx, beaconDB, util.NewBeaconBlock()).Block().HashTreeRoot()
require.NoError(t, err)
require.NoError(t, service.SaveState(ctx, r, beaconState))
// Should save both state and state summary.
@@ -43,7 +45,8 @@ func TestSaveState_HotStateCached(t *testing.T) {
require.NoError(t, beaconState.SetSlot(params.BeaconConfig().SlotsPerEpoch))
// Cache the state prior.
r := [32]byte{'a'}
r, err := util.SaveBlock(t, ctx, beaconDB, util.NewBeaconBlock()).Block().HashTreeRoot()
require.NoError(t, err)
service.hotStateCache.put(r, beaconState)
require.NoError(t, service.SaveState(ctx, r, beaconState))
@@ -61,7 +64,8 @@ func TestState_ForceCheckpoint_SavesStateToDatabase(t *testing.T) {
beaconState, _ := util.DeterministicGenesisState(t, 32)
require.NoError(t, beaconState.SetSlot(params.BeaconConfig().SlotsPerEpoch))
r := [32]byte{'a'}
r, err := util.SaveBlock(t, ctx, beaconDB, util.NewBeaconBlock()).Block().HashTreeRoot()
require.NoError(t, err)
svc.hotStateCache.put(r, beaconState)
require.Equal(t, false, beaconDB.HasState(ctx, r), "Database has state stored already")
@@ -80,7 +84,8 @@ func TestSaveState_Alreadyhas(t *testing.T) {
beaconState, _ := util.DeterministicGenesisState(t, 32)
require.NoError(t, beaconState.SetSlot(params.BeaconConfig().SlotsPerEpoch))
r := [32]byte{'A'}
r, err := util.SaveBlock(t, ctx, beaconDB, util.NewBeaconBlock()).Block().HashTreeRoot()
require.NoError(t, err)
// Pre cache the hot state.
service.hotStateCache.put(r, beaconState)
@@ -99,7 +104,8 @@ func TestSaveState_CanSaveOnEpochBoundary(t *testing.T) {
beaconState, _ := util.DeterministicGenesisState(t, 32)
require.NoError(t, beaconState.SetSlot(params.BeaconConfig().SlotsPerEpoch))
r := [32]byte{'A'}
r, err := util.SaveBlock(t, ctx, beaconDB, util.NewBeaconBlock()).Block().HashTreeRoot()
require.NoError(t, err)
require.NoError(t, service.saveStateByRoot(ctx, r, beaconState))
@@ -120,8 +126,10 @@ func TestSaveState_NoSaveNotEpochBoundary(t *testing.T) {
beaconState, _ := util.DeterministicGenesisState(t, 32)
require.NoError(t, beaconState.SetSlot(params.BeaconConfig().SlotsPerEpoch-1))
r := [32]byte{'A'}
r, err := util.SaveBlock(t, ctx, beaconDB, util.NewBeaconBlock()).Block().HashTreeRoot()
require.NoError(t, err)
b := util.NewBeaconBlock()
b.Block.Body.Graffiti = bytesutil.PadTo([]byte("foo"), 32)
util.SaveBlock(t, ctx, beaconDB, b)
gRoot, err := b.Block.HashTreeRoot()
require.NoError(t, err)
@@ -145,7 +153,8 @@ func TestSaveState_CanSaveHotStateToDB(t *testing.T) {
beaconState, _ := util.DeterministicGenesisState(t, 32)
require.NoError(t, beaconState.SetSlot(defaultHotStateDBInterval))
r := [32]byte{'A'}
r, err := util.SaveBlock(t, ctx, beaconDB, util.NewBeaconBlock()).Block().HashTreeRoot()
require.NoError(t, err)
require.NoError(t, service.saveStateByRoot(ctx, r, beaconState))
require.LogsContain(t, hook, "Saving hot state to DB")

View File

@@ -216,6 +216,7 @@ func TestValidateBeaconBlockPubSub_IsInCache(t *testing.T) {
ctx := context.Background()
beaconState, privKeys := util.DeterministicGenesisState(t, 100)
parentBlock := util.NewBeaconBlock()
util.SaveBlock(t, ctx, db, parentBlock)
bRoot, err := parentBlock.Block.HashTreeRoot()
require.NoError(t, err)
require.NoError(t, db.SaveState(ctx, beaconState, bRoot))

2
go.mod
View File

@@ -84,6 +84,7 @@ require (
go.etcd.io/bbolt v1.3.5
go.opencensus.io v0.23.0
go.uber.org/automaxprocs v1.3.0
go.uber.org/multierr v1.8.0
golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d
golang.org/x/exp v0.0.0-20220426173459-3bcf042a4bf5
golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4
@@ -221,7 +222,6 @@ require (
github.com/whyrusleeping/timecache v0.0.0-20160911033111-cfcb2f1abfee // indirect
github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 // indirect
go.uber.org/atomic v1.9.0 // indirect
go.uber.org/multierr v1.8.0 // indirect
go.uber.org/zap v1.21.0 // indirect
golang.org/x/net v0.0.0-20220624214902-1bab6f366d9e // indirect
golang.org/x/oauth2 v0.0.0-20220223155221-ee480838109b // indirect

View File

@@ -75,6 +75,9 @@ func handleConditionalExpression(exp *ast.BinaryExpr, pass *analysis.Pass) {
return
}
for _, iface := range selectedInterfaces {
if strings.HasPrefix(typeMap[identX].Type.String(), "func") {
return // Ignore functions as they are not interfaces.
}
xIsIface := strings.Contains(typeMap[identX].Type.String(), iface)
xIsNil := typeMap[identX].IsNil()
yisIface := strings.Contains(typeMap[identY].Type.String(), iface)
@@ -94,5 +97,6 @@ func handleConditionalExpression(exp *ast.BinaryExpr, pass *analysis.Pass) {
func reportFailure(pos token.Pos, pass *analysis.Pass) {
pass.Reportf(pos, "A single nilness check is being performed on an interface"+
", this check needs another accompanying nilness check on the underlying object for the interface.")
", this check needs another accompanying nilness check on the underlying object for the interface. "+
"Please use IsNil() if it is available")
}