Skip to content

sync blobs in initial-sync #12522

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

Merged
terencechain merged 16 commits into from
Jul 25, 2023
Merged

sync blobs in initial-sync #12522

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
bdbf0f9
deneb initial-sync
kasey Jun 10, 2023
49287cd
fix block unblinding bug mutating block encoding
kasey Jun 29, 2023
ebce6c7
fix missed usages of onBlockBatch
kasey Jun 30, 2023
9e1e675
fix lint errors
kasey Jun 30, 2023
aa6a0da
handle blob fetching in findFork code path
kasey Jul 12, 2023
2076738
fix to deal with changes from upstream
kasey Jul 12, 2023
f9216d1
refactor blob fetching for testability
kasey Jul 14, 2023
d0b4313
potuz feedback on wasteful slice alloc
kasey Jul 20, 2023
27a7fc7
small fixes and comments per Radek and Nishant
kasey Jul 20, 2023
953bd68
downscore peers for sending excess blobs
kasey Jul 20, 2023
e3d61a7
fix rebase bugs
kasey Jul 20, 2023
b814d46
fix bazel test build
kasey Jul 24, 2023
27d2eca
consume req before stream close to fix race
kasey Jul 24, 2023
bc45439
lint
kasey Jul 24, 2023
3de8fe7
lower bound block must have commitments
kasey Jul 25, 2023
7ee5e7f
regression test, slot with commits not lowest
kasey Jul 25, 2023
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
2 changes: 0 additions & 2 deletions beacon-chain/blockchain/error.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,6 @@ var (
errNilJustifiedCheckpoint = errors.New("nil justified checkpoint returned from state")
// errBlockDoesNotExist is returned when a block does not exist for a particular state summary.
errBlockDoesNotExist = errors.New("could not find block in DB")
// errWrongBlockCount is returned when the wrong number of blocks or block roots is used
errWrongBlockCount = errors.New("wrong number of blocks or block roots")
// errBlockNotFoundInCacheOrDB is returned when a block is not found in the cache or DB.
errBlockNotFoundInCacheOrDB = errors.New("block not found in cache or db")
// errWSBlockNotFound is returned when a block is not found in the WS cache or DB.
Expand Down
9 changes: 7 additions & 2 deletions beacon-chain/blockchain/log.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,7 @@ func logBlockSyncStatus(block interfaces.ReadOnlyBeaconBlock, blockRoot [32]byte
level := log.Logger.GetLevel()
if level >= logrus.DebugLevel {
parentRoot := block.ParentRoot()
log.WithFields(logrus.Fields{
lf := logrus.Fields{
"slot": block.Slot(),
"slotInEpoch": block.Slot() % params.BeaconConfig().SlotsPerEpoch,
"block": fmt.Sprintf("0x%s...", hex.EncodeToString(blockRoot[:])[:8]),
Expand All @@ -87,7 +87,12 @@ func logBlockSyncStatus(block interfaces.ReadOnlyBeaconBlock, blockRoot [32]byte
"sinceSlotStartTime": prysmTime.Now().Sub(startTime),
"chainServiceProcessedTime": prysmTime.Now().Sub(receivedTime),
"deposits": len(block.Body().Deposits()),
}).Debug("Synced new block")
}
commits, err := block.Body().BlobKzgCommitments()
if err == nil {
lf["commitments"] = len(commits)
}
log.WithFields(lf).Debug("Synced new block")
} else {
log.WithFields(logrus.Fields{
"slot": block.Slot(),
Expand Down
22 changes: 9 additions & 13 deletions beacon-chain/blockchain/process_block.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -151,19 +151,14 @@ func getStateVersionAndPayload(st state.BeaconState) (int, interfaces.ExecutionD
return preStateVersion, preStateHeader, nil
}

func (s *Service) onBlockBatch(ctx context.Context, blks []interfaces.ReadOnlySignedBeaconBlock,
blockRoots [][32]byte) error {
func (s *Service) onBlockBatch(ctx context.Context, blks []consensusblocks.ROBlock) error {
Copy link
Copy Markdown
Contributor

@potuz potuz Jul 17, 2023

Choose a reason for hiding this comment

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

this is not related to this PR, but I would prefer the type to signal that the block is signed.

Copy link
Copy Markdown
Collaborator

@terencechain terencechain Jul 17, 2023

Choose a reason for hiding this comment

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

maybe this can be another PR against develop if we decide to go with RO block route here

ctx, span := trace.StartSpan(ctx, "blockChain.onBlockBatch")
defer span.End()

if len(blks) == 0 || len(blockRoots) == 0 {
if len(blks) == 0 {
return errors.New("no blocks provided")
}

if len(blks) != len(blockRoots) {
return errWrongBlockCount
}

if err := consensusblocks.BeaconBlockIsNil(blks[0]); err != nil {
return invalidBlock{error: err}
}
Expand Down Expand Up @@ -213,7 +208,7 @@ func (s *Service) onBlockBatch(ctx context.Context, blks []interfaces.ReadOnlySi
}
// Save potential boundary states.
if slots.IsEpochStart(preState.Slot()) {
boundaries[blockRoots[i]] = preState.Copy()
boundaries[b.Root()] = preState.Copy()
}
jCheckpoints[i] = preState.CurrentJustifiedCheckpoint()
fCheckpoints[i] = preState.FinalizedCheckpoint()
Expand Down Expand Up @@ -246,11 +241,12 @@ func (s *Service) onBlockBatch(ctx context.Context, blks []interfaces.ReadOnlySi
pendingNodes := make([]*forkchoicetypes.BlockAndCheckpoints, len(blks))
var isValidPayload bool
for i, b := range blks {
root := b.Root()
isValidPayload, err = s.notifyNewPayload(ctx,
postVersionAndHeaders[i].version,
postVersionAndHeaders[i].header, b)
if err != nil {
return s.handleInvalidExecutionError(ctx, err, blockRoots[i], b.Block().ParentRoot())
return s.handleInvalidExecutionError(ctx, err, root, b.Block().ParentRoot())
}
if isValidPayload {
if err := s.validateMergeTransitionBlock(ctx, preVersionAndHeaders[i].version,
Expand All @@ -262,13 +258,13 @@ func (s *Service) onBlockBatch(ctx context.Context, blks []interfaces.ReadOnlySi
JustifiedCheckpoint: jCheckpoints[i],
FinalizedCheckpoint: fCheckpoints[i]}
pendingNodes[len(blks)-i-1] = args
if err := s.saveInitSyncBlock(ctx, blockRoots[i], b); err != nil {
if err := s.saveInitSyncBlock(ctx, root, b); err != nil {
Copy link
Copy Markdown
Contributor

@potuz potuz Jul 17, 2023

Choose a reason for hiding this comment

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

This function can now be modified to not take the root and use b itself.

Copy link
Copy Markdown
Contributor

@nisdas nisdas Jul 18, 2023

Choose a reason for hiding this comment

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

+1 , although better to have it as a follow up. Also on another note the ROBlock should go into develop directly and we implement all these changes there

Copy link
Copy Markdown
Collaborator Author

@kasey kasey Jul 19, 2023

Choose a reason for hiding this comment

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

It's in develop,Nishant: https://github.com/prysmaticlabs/prysm/blob/develop/consensus-types/blocks/roblock.go

Copy link
Copy Markdown
Contributor

@nisdas nisdas Jul 19, 2023

Choose a reason for hiding this comment

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

I guess more on changing the block type from SignedBeaconBlock to RoBlock across initial sync

tracing.AnnotateError(span, err)
return err
}
if err := s.cfg.BeaconDB.SaveStateSummary(ctx, &ethpb.StateSummary{
Slot: b.Block().Slot(),
Root: blockRoots[i][:],
Root: root[:],
}); err != nil {
tracing.AnnotateError(span, err)
return err
Expand All @@ -292,8 +288,9 @@ func (s *Service) onBlockBatch(ctx context.Context, blks []interfaces.ReadOnlySi
return err
}
}
lastB := blks[len(blks)-1]
lastBR := lastB.Root()
// Also saves the last post state which to be used as pre state for the next batch.
lastBR := blockRoots[len(blks)-1]
if err := s.cfg.StateGen.SaveState(ctx, lastBR, preState); err != nil {
return err
}
Expand All @@ -311,7 +308,6 @@ func (s *Service) onBlockBatch(ctx context.Context, blks []interfaces.ReadOnlySi
return errors.Wrap(err, "could not set optimistic block to valid")
}
}
lastB := blks[len(blks)-1]
arg := &notifyForkchoiceUpdateArg{
headState: preState,
headRoot: lastBR,
Expand Down
30 changes: 14 additions & 16 deletions beacon-chain/blockchain/process_block_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,8 +49,7 @@ func TestStore_OnBlockBatch(t *testing.T) {
require.NoError(t, service.saveGenesisData(ctx, st))
bState := st.Copy()

var blks []interfaces.ReadOnlySignedBeaconBlock
var blkRoots [][32]byte
var blks []consensusblocks.ROBlock
for i := 0; i < 97; i++ {
b, err := util.GenerateFullBlock(bState, keys, util.DefaultBlockGenConfig(), primitives.Slot(i))
require.NoError(t, err)
Expand All @@ -63,16 +62,15 @@ func TestStore_OnBlockBatch(t *testing.T) {
require.NoError(t, service.saveInitSyncBlock(ctx, root, wsb))
wsb, err = consensusblocks.NewSignedBeaconBlock(b)
require.NoError(t, err)
blks = append(blks, wsb)
blkRoots = append(blkRoots, root)
rwsb, err := consensusblocks.NewROBlock(wsb)
require.NoError(t, err)
blks = append(blks, rwsb)
}
err := service.onBlockBatch(ctx, blks, blkRoots[1:])
require.ErrorIs(t, errWrongBlockCount, err)
err = service.onBlockBatch(ctx, blks, blkRoots)
err := service.onBlockBatch(ctx, blks)
require.NoError(t, err)
jcp := service.CurrentJustifiedCheckpt()
jroot := bytesutil.ToBytes32(jcp.Root)
require.Equal(t, blkRoots[63], jroot)
require.Equal(t, blks[63].Root(), jroot)
require.Equal(t, primitives.Epoch(2), service.cfg.ForkChoiceStore.JustifiedCheckpoint().Epoch)
}

Expand All @@ -84,8 +82,7 @@ func TestStore_OnBlockBatch_NotifyNewPayload(t *testing.T) {
require.NoError(t, service.saveGenesisData(ctx, st))
bState := st.Copy()

var blks []interfaces.ReadOnlySignedBeaconBlock
var blkRoots [][32]byte
var blks []consensusblocks.ROBlock
blkCount := 4
for i := 0; i <= blkCount; i++ {
b, err := util.GenerateFullBlock(bState, keys, util.DefaultBlockGenConfig(), primitives.Slot(i))
Expand All @@ -94,13 +91,12 @@ func TestStore_OnBlockBatch_NotifyNewPayload(t *testing.T) {
require.NoError(t, err)
bState, err = transition.ExecuteStateTransition(ctx, bState, wsb)
require.NoError(t, err)
root, err := b.Block.HashTreeRoot()
rwsb, err := consensusblocks.NewROBlock(wsb)
require.NoError(t, err)
require.NoError(t, service.saveInitSyncBlock(ctx, root, wsb))
blks = append(blks, wsb)
blkRoots = append(blkRoots, root)
require.NoError(t, service.saveInitSyncBlock(ctx, rwsb.Root(), wsb))
blks = append(blks, rwsb)
}
require.NoError(t, service.onBlockBatch(ctx, blks, blkRoots))
require.NoError(t, service.onBlockBatch(ctx, blks))
}

func TestCachedPreState_CanGetFromStateSummary(t *testing.T) {
Expand Down Expand Up @@ -1933,8 +1929,10 @@ func TestNoViableHead_Reboot(t *testing.T) {
require.NoError(t, err)
root, err = b.Block.HashTreeRoot()
require.NoError(t, err)
rwsb, err := consensusblocks.NewROBlock(wsb)
require.NoError(t, err)
// We use onBlockBatch here because the valid chain is missing in forkchoice
require.NoError(t, service.onBlockBatch(ctx, []interfaces.ReadOnlySignedBeaconBlock{wsb}, [][32]byte{root}))
require.NoError(t, service.onBlockBatch(ctx, []consensusblocks.ROBlock{rwsb}))
// Check that the head is now VALID and the node is not optimistic
require.Equal(t, genesisRoot, service.ensureRootNotZeros(service.cfg.ForkChoiceStore.CachedHeadRoot()))
headRoot, err = service.HeadRoot(ctx)
Expand Down
13 changes: 7 additions & 6 deletions beacon-chain/blockchain/receive_block.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import (
"github.com/prysmaticlabs/prysm/v4/beacon-chain/core/transition"
"github.com/prysmaticlabs/prysm/v4/beacon-chain/state"
"github.com/prysmaticlabs/prysm/v4/config/features"
"github.com/prysmaticlabs/prysm/v4/consensus-types/blocks"
"github.com/prysmaticlabs/prysm/v4/consensus-types/interfaces"
"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
"github.com/prysmaticlabs/prysm/v4/encoding/bytesutil"
Expand All @@ -30,7 +31,7 @@ var epochsSinceFinalitySaveHotStateDB = primitives.Epoch(100)
// BlockReceiver interface defines the methods of chain service for receiving and processing new blocks.
type BlockReceiver interface {
ReceiveBlock(ctx context.Context, block interfaces.ReadOnlySignedBeaconBlock, blockRoot [32]byte) error
ReceiveBlockBatch(ctx context.Context, blocks []interfaces.ReadOnlySignedBeaconBlock, blkRoots [][32]byte) error
ReceiveBlockBatch(ctx context.Context, blocks []blocks.ROBlock) error
HasBlock(ctx context.Context, root [32]byte) bool
RecentBlockSlot(root [32]byte) (primitives.Slot, error)
}
Expand Down Expand Up @@ -150,26 +151,26 @@ func (s *Service) ReceiveBlock(ctx context.Context, block interfaces.ReadOnlySig
// ReceiveBlockBatch processes the whole block batch at once, assuming the block batch is linear ,transitioning
// the state, performing batch verification of all collected signatures and then performing the appropriate
// actions for a block post-transition.
func (s *Service) ReceiveBlockBatch(ctx context.Context, blocks []interfaces.ReadOnlySignedBeaconBlock, blkRoots [][32]byte) error {
func (s *Service) ReceiveBlockBatch(ctx context.Context, blocks []blocks.ROBlock) error {
ctx, span := trace.StartSpan(ctx, "blockChain.ReceiveBlockBatch")
defer span.End()

s.cfg.ForkChoiceStore.Lock()
defer s.cfg.ForkChoiceStore.Unlock()

// Apply state transition on the incoming newly received block batches, one by one.
if err := s.onBlockBatch(ctx, blocks, blkRoots); err != nil {
if err := s.onBlockBatch(ctx, blocks); err != nil {
err := errors.Wrap(err, "could not process block in batch")
tracing.AnnotateError(span, err)
return err
}

for i, b := range blocks {
for _, b := range blocks {
blockCopy, err := b.Copy()
if err != nil {
return err
}
optimistic, err := s.cfg.ForkChoiceStore.IsOptimistic(blkRoots[i])
optimistic, err := s.cfg.ForkChoiceStore.IsOptimistic(b.Root())
if err != nil {
log.WithError(err).Error("Could not check if block is optimistic")
optimistic = true
Expand All @@ -179,7 +180,7 @@ func (s *Service) ReceiveBlockBatch(ctx context.Context, blocks []interfaces.Rea
Type: statefeed.BlockProcessed,
Data: &statefeed.BlockProcessedData{
Slot: blockCopy.Block().Slot(),
BlockRoot: blkRoots[i],
BlockRoot: b.Root(),
SignedBlock: blockCopy,
Verified: true,
Optimistic: optimistic,
Expand Down
9 changes: 3 additions & 6 deletions beacon-chain/blockchain/receive_block_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,6 @@ import (
"github.com/prysmaticlabs/prysm/v4/beacon-chain/operations/voluntaryexits"
"github.com/prysmaticlabs/prysm/v4/config/params"
"github.com/prysmaticlabs/prysm/v4/consensus-types/blocks"
"github.com/prysmaticlabs/prysm/v4/consensus-types/interfaces"
"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
"github.com/prysmaticlabs/prysm/v4/encoding/bytesutil"
ethpb "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
Expand Down Expand Up @@ -228,13 +227,11 @@ func TestService_ReceiveBlockBatch(t *testing.T) {
s, _ := minimalTestService(t, WithStateNotifier(&blockchainTesting.MockStateNotifier{RecordEvents: true}))
err := s.saveGenesisData(ctx, genesis)
require.NoError(t, err)
root, err := tt.args.block.Block.HashTreeRoot()
require.NoError(t, err)
wsb, err := blocks.NewSignedBeaconBlock(tt.args.block)
require.NoError(t, err)
blks := []interfaces.ReadOnlySignedBeaconBlock{wsb}
roots := [][32]byte{root}
err = s.ReceiveBlockBatch(ctx, blks, roots)
rwsb, err := blocks.NewROBlock(wsb)
require.NoError(t, err)
err = s.ReceiveBlockBatch(ctx, []blocks.ROBlock{rwsb})
if tt.wantedErr != "" {
assert.ErrorContains(t, tt.wantedErr, err)
} else {
Expand Down
1 change: 1 addition & 0 deletions beacon-chain/blockchain/testing/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ go_library(
"//beacon-chain/state/state-native:go_default_library",
"//config/fieldparams:go_default_library",
"//config/params:go_default_library",
"//consensus-types/blocks:go_default_library",
"//consensus-types/interfaces:go_default_library",
"//consensus-types/primitives:go_default_library",
"//encoding/bytesutil:go_default_library",
Expand Down
3 changes: 2 additions & 1 deletion beacon-chain/blockchain/testing/mock.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import (
state_native "github.com/prysmaticlabs/prysm/v4/beacon-chain/state/state-native"
fieldparams "github.com/prysmaticlabs/prysm/v4/config/fieldparams"
"github.com/prysmaticlabs/prysm/v4/config/params"
"github.com/prysmaticlabs/prysm/v4/consensus-types/blocks"
"github.com/prysmaticlabs/prysm/v4/consensus-types/interfaces"
"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
"github.com/prysmaticlabs/prysm/v4/encoding/bytesutil"
Expand Down Expand Up @@ -204,7 +205,7 @@ func (s *ChainService) ReceiveBlockInitialSync(ctx context.Context, block interf
}

// ReceiveBlockBatch processes blocks in batches from initial-sync.
func (s *ChainService) ReceiveBlockBatch(ctx context.Context, blks []interfaces.ReadOnlySignedBeaconBlock, _ [][32]byte) error {
func (s *ChainService) ReceiveBlockBatch(ctx context.Context, blks []blocks.ROBlock) error {
if s.State == nil {
return ErrNilState
}
Expand Down
1 change: 1 addition & 0 deletions beacon-chain/execution/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,7 @@ go_library(
"@com_github_sirupsen_logrus//:go_default_library",
"@io_k8s_client_go//tools/cache:go_default_library",
"@io_opencensus_go//trace:go_default_library",
"@org_golang_google_protobuf//proto:go_default_library",
],
)

Expand Down
Loading