Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Prevent Reprocessing of a Block From Our Pending Queue #9904

Merged
merged 6 commits into from
Nov 17, 2021
Merged

Prevent Reprocessing of a Block From Our Pending Queue #9904

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
1c70a7e
fix bugs
nisdas Nov 15, 2021
43adf1d
Merge branch 'develop' of https://github.com/prysmaticlabs/geth-shard…
nisdas Nov 15, 2021
493ddca
test
nisdas Nov 16, 2021
bfcfd7e
Merge branch 'develop' into fixPendingBlockBug
nisdas Nov 17, 2021
57bd133
raul's review
nisdas Nov 17, 2021
ad1d85a
Merge branch 'develop' into fixPendingBlockBug
nisdas Nov 17, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 22 additions & 7 deletions beacon-chain/sync/pending_blocks_queue.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,16 +88,29 @@ func (s *Service) processPendingBlocks(ctx context.Context) error {
continue
}

s.pendingQueueLock.RLock()
inPendingQueue := s.seenPendingBlocks[bytesutil.ToBytes32(b.Block().ParentRoot())]
s.pendingQueueLock.RUnlock()

blkRoot, err := b.Block().HashTreeRoot()
if err != nil {
tracing.AnnotateError(span, err)
span.End()
return err
}
inDB := s.cfg.beaconDB.HasBlock(ctx, blkRoot)
// No need to process the same block twice.
if inDB {
s.pendingQueueLock.Lock()
if err := s.deleteBlockFromPendingQueue(slot, b, blkRoot); err != nil {
s.pendingQueueLock.Unlock()
return err
}
s.pendingQueueLock.Unlock()
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

you need to release the lock if there's an error in the line above as well

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah , very nice catch. We use the same pattern in a few places, will fix them.

span.End()
continue
}

s.pendingQueueLock.RLock()
inPendingQueue := s.seenPendingBlocks[bytesutil.ToBytes32(b.Block().ParentRoot())]
s.pendingQueueLock.RUnlock()

parentIsBad := s.hasBadBlock(bytesutil.ToBytes32(b.Block().ParentRoot()))
blockIsBad := s.hasBadBlock(blkRoot)
// Check if parent is a bad block.
Expand All @@ -117,12 +130,12 @@ func (s *Service) processPendingBlocks(ctx context.Context) error {
continue
}

inDB := s.cfg.beaconDB.HasBlock(ctx, bytesutil.ToBytes32(b.Block().ParentRoot()))
parentInDb := s.cfg.beaconDB.HasBlock(ctx, bytesutil.ToBytes32(b.Block().ParentRoot()))
hasPeer := len(pids) != 0

// Only request for missing parent block if it's not in beaconDB, not in pending cache
// and has peer in the peer list.
if !inPendingQueue && !inDB && hasPeer {
if !inPendingQueue && !parentInDb && hasPeer {
log.WithFields(logrus.Fields{
"currentSlot": b.Block().Slot(),
"parentRoot": hex.EncodeToString(bytesutil.Trunc(b.Block().ParentRoot())),
Expand All @@ -133,7 +146,7 @@ func (s *Service) processPendingBlocks(ctx context.Context) error {
continue
}

if !inDB {
if !parentInDb {
span.End()
continue
}
Expand Down Expand Up @@ -167,6 +180,7 @@ func (s *Service) processPendingBlocks(ctx context.Context) error {

s.pendingQueueLock.Lock()
if err := s.deleteBlockFromPendingQueue(slot, b, blkRoot); err != nil {
s.pendingQueueLock.Unlock()
return err
}
s.pendingQueueLock.Unlock()
Expand Down Expand Up @@ -321,6 +335,7 @@ func (s *Service) deleteBlockFromPendingQueue(slot types.Slot, b block.SignedBea
}
if len(newBlks) == 0 {
s.slotToPendingBlocks.Delete(slotToCacheKey(slot))
delete(s.seenPendingBlocks, r)
return nil
}

Expand Down
56 changes: 50 additions & 6 deletions beacon-chain/sync/pending_blocks_queue_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,9 +84,13 @@ func TestRegularSyncBeaconBlockSubscriber_ProcessPendingBlocks1(t *testing.T) {
require.NoError(t, r.insertBlockToPendingQueue(b1.Block.Slot, wrapper.WrappedPhase0SignedBeaconBlock(b1), b1Root))
require.NoError(t, r.cfg.beaconDB.SaveBlock(context.Background(), wrapper.WrappedPhase0SignedBeaconBlock(b1)))

// Insert bad b1 in the cache to verify the good one doesn't get replaced.
require.NoError(t, r.insertBlockToPendingQueue(b1.Block.Slot, wrapper.WrappedPhase0SignedBeaconBlock(util.NewBeaconBlock()), [32]byte{}))
nBlock := util.NewBeaconBlock()
nBlock.Block.Slot = b1.Block.Slot
nRoot, err := nBlock.Block.HashTreeRoot()
require.NoError(t, err)

// Insert bad b1 in the cache to verify the good one doesn't get replaced.
require.NoError(t, r.insertBlockToPendingQueue(nBlock.Block.Slot, wrapper.WrappedPhase0SignedBeaconBlock(nBlock), nRoot))
require.NoError(t, r.processPendingBlocks(context.Background())) // Marks a block as bad
require.NoError(t, r.processPendingBlocks(context.Background())) // Bad block removed on second run

Expand Down Expand Up @@ -140,6 +144,46 @@ func TestRegularSync_InsertDuplicateBlocks(t *testing.T) {

}

func TestRegularSyncBeaconBlockSubscriber_DoNotReprocessBlock(t *testing.T) {
db := dbtest.SetupDB(t)

p1 := p2ptest.NewTestP2P(t)
r := &Service{
cfg: &config{
p2p: p1,
beaconDB: db,
chain: &mock.ChainService{
FinalizedCheckPoint: &ethpb.Checkpoint{
Epoch: 0,
},
},
stateGen: stategen.New(db),
},
slotToPendingBlocks: gcache.New(time.Second, 2*time.Second),
seenPendingBlocks: make(map[[32]byte]bool),
}
r.initCaches()

b0 := util.NewBeaconBlock()
require.NoError(t, r.cfg.beaconDB.SaveBlock(context.Background(), wrapper.WrappedPhase0SignedBeaconBlock(b0)))
b0Root, err := b0.Block.HashTreeRoot()
require.NoError(t, err)
b3 := util.NewBeaconBlock()
b3.Block.Slot = 3
b3.Block.ParentRoot = b0Root[:]
b3Root, err := b3.Block.HashTreeRoot()
require.NoError(t, err)

require.NoError(t, r.cfg.beaconDB.SaveBlock(context.Background(), wrapper.WrappedPhase0SignedBeaconBlock(b3)))

// Add b3 to the cache
require.NoError(t, r.insertBlockToPendingQueue(b3.Block.Slot, wrapper.WrappedPhase0SignedBeaconBlock(b3), b3Root))

require.NoError(t, r.processPendingBlocks(context.Background()))
assert.Equal(t, 0, len(r.slotToPendingBlocks.Items()), "Incorrect size for slot to pending blocks cache")
assert.Equal(t, 0, len(r.seenPendingBlocks), "Incorrect size for seen pending block")
}

// /- b1 - b2 - b5
// b0
// \- b3 - b4
Expand Down Expand Up @@ -237,7 +281,7 @@ func TestRegularSyncBeaconBlockSubscriber_ProcessPendingBlocks_2Chains(t *testin
require.NoError(t, r.processPendingBlocks(context.Background())) // Bad block removed on second run

assert.Equal(t, 1, len(r.slotToPendingBlocks.Items()), "Incorrect size for slot to pending blocks cache")
assert.Equal(t, 3, len(r.seenPendingBlocks), "Incorrect size for seen pending block")
assert.Equal(t, 1, len(r.seenPendingBlocks), "Incorrect size for seen pending block")

// Add b2 to the cache
require.NoError(t, r.insertBlockToPendingQueue(b2.Block.Slot, wrapper.WrappedPhase0SignedBeaconBlock(b2), b2Root))
Expand All @@ -248,7 +292,7 @@ func TestRegularSyncBeaconBlockSubscriber_ProcessPendingBlocks_2Chains(t *testin
require.NoError(t, r.processPendingBlocks(context.Background())) // Bad block removed on second run

assert.Equal(t, 0, len(r.slotToPendingBlocks.Items()), "Incorrect size for slot to pending blocks cache")
assert.Equal(t, 4, len(r.seenPendingBlocks), "Incorrect size for seen pending block")
assert.Equal(t, 0, len(r.seenPendingBlocks), "Incorrect size for seen pending block")
}

func TestRegularSyncBeaconBlockSubscriber_PruneOldPendingBlocks(t *testing.T) {
Expand Down Expand Up @@ -318,7 +362,7 @@ func TestRegularSyncBeaconBlockSubscriber_PruneOldPendingBlocks(t *testing.T) {

require.NoError(t, r.processPendingBlocks(context.Background()))
assert.Equal(t, 0, len(r.slotToPendingBlocks.Items()), "Incorrect size for slot to pending blocks cache")
assert.Equal(t, 4, len(r.seenPendingBlocks), "Incorrect size for seen pending block")
assert.Equal(t, 0, len(r.seenPendingBlocks), "Incorrect size for seen pending block")
}

func TestService_sortedPendingSlots(t *testing.T) {
Expand Down Expand Up @@ -429,7 +473,7 @@ func TestService_BatchRootRequest(t *testing.T) {
assert.Equal(t, 4, len(r.seenPendingBlocks), "Incorrect size for seen pending block")
}

func TestService_AddPeningBlockToQueueOverMax(t *testing.T) {
func TestService_AddPendingBlockToQueueOverMax(t *testing.T) {
r := &Service{
slotToPendingBlocks: gcache.New(time.Second, 2*time.Second),
seenPendingBlocks: make(map[[32]byte]bool),
Expand Down