diff --git a/beacon-chain/sync/pending_blocks_queue.go b/beacon-chain/sync/pending_blocks_queue.go index 113691102a81..de9a7d637268 100644 --- a/beacon-chain/sync/pending_blocks_queue.go +++ b/beacon-chain/sync/pending_blocks_queue.go @@ -34,6 +34,7 @@ var processPendingBlocksPeriod = slots.DivideSlotBy(3 /* times per slot */) const maxPeerRequest = 50 const numOfTries = 5 const maxBlocksPerSlot = 3 +const maxParentFetchesPerSlot = 4 // processes pending blocks queue on every processPendingBlocksPeriod func (s *Service) processPendingBlocksQueue() { @@ -364,6 +365,28 @@ func (s *Service) checkIfBlockIsBad( return true, nil } +func (s *Service) fetchMissingParent(root [32]byte) { + currentSlot := s.cfg.clock.CurrentSlot() + + s.parentFetchLock.Lock() + if s.parentFetchRoots == nil || s.parentFetchSlot != currentSlot { + s.parentFetchSlot = currentSlot + s.parentFetchRoots = make(map[[32]byte]bool, maxParentFetchesPerSlot) + } + if s.parentFetchRoots[root] || len(s.parentFetchRoots) >= maxParentFetchesPerSlot { + s.parentFetchLock.Unlock() + return + } + s.parentFetchRoots[root] = true + s.parentFetchLock.Unlock() + + go func() { + if err := s.sendBatchRootRequest(s.ctx, [][32]byte{root}, rand.NewGenerator()); err != nil { + log.WithError(err).WithField("root", fmt.Sprintf("%#x", root)).Debug("Failed to send batch root request") + } + }() +} + func (s *Service) sendBatchRootRequest(ctx context.Context, roots [][32]byte, randGen *rand.Rand) error { ctx, span := prysmTrace.StartSpan(ctx, "sendBatchRootRequest") defer span.End() diff --git a/beacon-chain/sync/pending_blocks_queue_test.go b/beacon-chain/sync/pending_blocks_queue_test.go index b63e3959e751..4abb728f12e2 100644 --- a/beacon-chain/sync/pending_blocks_queue_test.go +++ b/beacon-chain/sync/pending_blocks_queue_test.go @@ -973,3 +973,38 @@ func TestExpirationCache_PruneOldBlocksCorrectly(t *testing.T) { assert.Equal(t, false, r.seenPendingBlocks[b2Root]) assert.Equal(t, 0, len(r.pendingBlocksInCache(1))) } + +func TestService_fetchMissingParent_DedupsAndCapsPerSlot(t *testing.T) { + chain := &mock.ChainService{ + FinalizedCheckPoint: ðpb.Checkpoint{Epoch: 0, Root: make([]byte, 32)}, + ValidatorsRoot: [32]byte{}, + Genesis: time.Now(), + } + newService := func() *Service { + return &Service{ + ctx: t.Context(), + cfg: &config{ + p2p: p2ptest.NewTestP2P(t), + chain: chain, + clock: startup.NewClock(chain.Genesis, chain.ValidatorsRoot), + }, + seenPendingBlocks: make(map[[32]byte]bool), + } + } + + t.Run("same root is only fetched once", func(t *testing.T) { + r := newService() + for range 128 { + r.fetchMissingParent([32]byte{'a'}) + } + require.Equal(t, 1, len(r.parentFetchRoots)) + }) + + t.Run("distinct roots are capped", func(t *testing.T) { + r := newService() + for i := range 128 { + r.fetchMissingParent([32]byte{byte(i)}) + } + require.Equal(t, maxParentFetchesPerSlot, len(r.parentFetchRoots)) + }) +} diff --git a/beacon-chain/sync/service.go b/beacon-chain/sync/service.go index 8dfe7a4c501e..400e2db62efb 100644 --- a/beacon-chain/sync/service.go +++ b/beacon-chain/sync/service.go @@ -150,6 +150,9 @@ type Service struct { subHandler *subTopicHandler pendingAttsLock sync.RWMutex pendingQueueLock sync.RWMutex + parentFetchLock sync.Mutex + parentFetchSlot primitives.Slot + parentFetchRoots map[[32]byte]bool chainStarted *atomic.Bool validateBlockLock sync.RWMutex rateLimiter *limiter diff --git a/beacon-chain/sync/validate_data_column.go b/beacon-chain/sync/validate_data_column.go index 6ab6dbd948c7..5af60794fa10 100644 --- a/beacon-chain/sync/validate_data_column.go +++ b/beacon-chain/sync/validate_data_column.go @@ -16,7 +16,6 @@ import ( "github.com/OffchainLabs/prysm/v7/consensus-types/blocks" "github.com/OffchainLabs/prysm/v7/consensus-types/primitives" "github.com/OffchainLabs/prysm/v7/container/slice" - "github.com/OffchainLabs/prysm/v7/crypto/rand" "github.com/OffchainLabs/prysm/v7/encoding/bytesutil" eth "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1" "github.com/OffchainLabs/prysm/v7/runtime/logging" @@ -183,19 +182,12 @@ func (s *Service) validateDataColumnFulu( // [IGNORE] The sidecar's block's parent (defined by `block_header.parent_root`) has been seen (via gossip or non-gossip sources // (a client MAY queue sidecars for processing once the parent block is retrieved). if err := verifier.SidecarParentSeen(s.hasBadBlock); err != nil { - go func() { - customCtx := context.Background() - parentRoot, err := roDataColumn.ParentRoot() - if err != nil { - log.WithError(err).WithFields(logging.DataColumnFields(roDataColumn)).Debug("Failed to get parent root for batch root request") - return - } - roots := [][fieldparams.RootLength]byte{parentRoot} - randGenerator := rand.NewGenerator() - if reqErr := s.sendBatchRootRequest(customCtx, roots, randGenerator); reqErr != nil { - log.WithError(reqErr).WithFields(logging.DataColumnFields(roDataColumn)).Debug("Failed to send batch root request") - } - }() + parentRoot, rootErr := roDataColumn.ParentRoot() + if rootErr != nil { + log.WithError(rootErr).WithFields(logging.DataColumnFields(roDataColumn)).Debug("Failed to get parent root for batch root request") + } else { + s.fetchMissingParent(parentRoot) + } return blocks.VerifiedRODataColumn{}, ignoreValidation(err) } diff --git a/changelog/terence_bound_column_parent_root_fetch.md b/changelog/terence_bound_column_parent_root_fetch.md new file mode 100644 index 000000000000..cab453547b80 --- /dev/null +++ b/changelog/terence_bound_column_parent_root_fetch.md @@ -0,0 +1,3 @@ +### Fixed + +- Dedupe and cap the parent block fetches triggered by unknown-parent data column sidecars.