Skip to content

cl/phase1/network/services: extract shared pending-job queue for the Gloas services #22177

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

Draft
yperbasis wants to merge 3 commits into
base: main
Choose a base branch
from
Draft

cl/phase1/network/services: extract shared pending-job queue for the Gloas services #22177

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f76f724
cl/phase1/network/services: pin execution payload bid pending-queue b…
yperbasis Jul 2, 2026
2b14eaa
cl/phase1/network/services: extract shared pendingJobQueue for Gloas …
yperbasis Jul 2, 2026
174d4d3
cl/phase1/network/services: dedupe isLocalValidatorProposer into loca…
yperbasis Jul 2, 2026
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
45 changes: 2 additions & 43 deletions cl/phase1/network/services/aggregate_and_proof_service.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,9 +77,7 @@ type aggregateAndProofServiceImpl struct {
batchSignatureVerifier *BatchSignatureVerifier
seenAggreatorIndexes *lru.Cache[seenAggregateIndex, struct{}]
validatorParams *validator_params.ValidatorParams

// Cached proposer indices per epoch (for current epoch check)
proposerIndicesCache *lru.Cache[uint64, []uint64]
localProposerChecker

// set of aggregates that are scheduled for later processing
aggregatesScheduledForLaterExecution sync.Map
Expand All @@ -99,10 +97,6 @@ func NewAggregateAndProofService(
if err != nil {
panic(err)
}
proposerIndicesCache, err := lru.New[uint64, []uint64]("proposerIndices", 3)
if err != nil {
panic(err)
}
a := &aggregateAndProofServiceImpl{
syncedDataManager: syncedDataManager,
forkchoiceStore: forkchoiceStore,
Expand All @@ -112,7 +106,7 @@ func NewAggregateAndProofService(
batchSignatureVerifier: batchSignatureVerifier,
seenAggreatorIndexes: seenAggCache,
validatorParams: validatorParams,
proposerIndicesCache: proposerIndicesCache,
localProposerChecker: newLocalProposerChecker(beaconCfg),
}
go a.loop(ctx)
return a
Expand All @@ -126,41 +120,6 @@ func (a *aggregateAndProofServiceImpl) IsMyGossipMessage(name string) bool {
return name == gossip.TopicNameBeaconAggregateAndProof
}

// isLocalValidatorProposer checks if any local validator is a proposer in the current or next epoch.
// For current epoch: uses cached GetBeaconProposerIndices.
// For next epoch: uses GetProposerLookahead() (Fulu+) or always returns true (pre-Fulu).
func (a *aggregateAndProofServiceImpl) isLocalValidatorProposer(headState *state.CachingBeaconState, currentEpoch uint64, localValidators []uint64) bool {
if headState.Version() < clparams.FuluVersion {
return true
}
// Check current epoch using cached proposer indices
currentProposers, ok := a.proposerIndicesCache.Get(currentEpoch)
if !ok {
var err error
currentProposers, err = headState.GetBeaconProposerIndices(currentEpoch)
if err == nil {
a.proposerIndicesCache.Add(currentEpoch, currentProposers)
}
}
for _, validatorIndex := range localValidators {
if slices.Contains(currentProposers, validatorIndex) {
return true
}
}

// For Fulu+, use the efficient proposer lookahead for next epoch
lookahead := headState.GetProposerLookahead()
// The lookahead contains proposers for current and next epoch, skip current epoch slots
slotsPerEpoch := int(a.beaconCfg.SlotsPerEpoch)
for i := slotsPerEpoch; i < lookahead.Length(); i++ {
proposerIndex := lookahead.Get(i)
if slices.Contains(localValidators, proposerIndex) {
return true
}
}
return false
}

func (a *aggregateAndProofServiceImpl) DecodeGossipMessage(pid peer.ID, data []byte, version clparams.StateVersion) (*SignedAggregateAndProofForGossip, error) {
obj := &SignedAggregateAndProofForGossip{
Receiver: &sentinelproto.Peer{Pid: pid.String()},
Expand Down
158 changes: 41 additions & 117 deletions cl/phase1/network/services/execution_payload_bid_service.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,6 @@ import (
"context"
"errors"
"fmt"
"sync"
"sync/atomic"
"time"

"github.com/erigontech/erigon/cl/beacon/beaconevents"
Expand Down Expand Up @@ -52,12 +50,6 @@ type pendingBidKey struct {
slot uint64
}

// pendingBidJob represents a pending bid waiting for proposer preferences to arrive.
type pendingBidJob struct {
msg *cltypes.SignedExecutionPayloadBid
creationTime time.Time
}

var errBidDependencyUnavailable = fmt.Errorf("%w: bid dependency unavailable", ErrIgnore)

const (
Expand All @@ -81,9 +73,7 @@ type executionPayloadBidService struct {
seenCache *lru.Cache[seenBidKey, struct{}]

// Pending bids waiting for proposer preferences
pendingBids sync.Map // pendingBidKey -> *pendingBidJob
pendingCount atomic.Int32
pendingCond *sync.Cond
pending *pendingJobQueue[pendingBidKey, *cltypes.SignedExecutionPayloadBid]
}

// NewExecutionPayloadBidService creates a new execution payload bid gossip service.
Expand All @@ -109,12 +99,21 @@ func NewExecutionPayloadBidService(
epbsPool: epbsPool,
emitters: emitters,
seenCache: seenCache,
pendingCond: sync.NewCond(&sync.Mutex{}),
}
go s.loop(ctx)
s.pending = s.newPendingQueue()
go s.pending.loop(ctx)
return s
}

func (s *executionPayloadBidService) newPendingQueue() *pendingJobQueue[pendingBidKey, *cltypes.SignedExecutionPayloadBid] {
return newPendingJobQueue(maxPendingBids, pendingBidExpiry, pendingBidCheckInterval,
s.tryProcessPendingBid,
func(key pendingBidKey) {
log.Trace("Pending execution payload bid expired",
"slot", key.slot, "builderIndex", key.builderIndex)
})
}

func (s *executionPayloadBidService) Names() []string {
return []string{gossip.TopicNameExecutionPayloadBid}
}
Expand Down Expand Up @@ -334,119 +333,44 @@ func (s *executionPayloadBidService) validateAndStoreBid(

// queuePendingBid adds a bid to the pending queue for later processing when preferences arrive.
func (s *executionPayloadBidService) queuePendingBid(msg *cltypes.SignedExecutionPayloadBid) {
if s.pendingCount.Add(1) > maxPendingBids {
s.pendingCount.Add(-1)
return
}

key := pendingBidKey{
s.pending.enqueue(pendingBidKey{
builderIndex: msg.Message.BuilderIndex,
slot: msg.Message.Slot,
}

if _, loaded := s.pendingBids.LoadOrStore(key, &pendingBidJob{
msg: msg,
creationTime: time.Now(),
}); loaded {
s.pendingCount.Add(-1)
} else {
s.pendingCond.L.Lock()
s.pendingCond.Signal()
s.pendingCond.L.Unlock()
}
}, msg)
}

// loop is the background goroutine that processes pending bids.
func (s *executionPayloadBidService) loop(ctx context.Context) {
// Wake any blocked Wait() on context cancellation to prevent deadlock.
go func() {
<-ctx.Done()
s.pendingCond.L.Lock()
s.pendingCond.Broadcast()
s.pendingCond.L.Unlock()
}()

for {
// Wait until there are pending bids
s.pendingCond.L.Lock()
for s.pendingCount.Load() == 0 {
select {
case <-ctx.Done():
s.pendingCond.L.Unlock()
return
default:
}
s.pendingCond.Wait()
}
s.pendingCond.L.Unlock()

// Poll until all pending bids are processed
ticker := time.NewTicker(pendingBidCheckInterval)
for s.pendingCount.Load() > 0 {
select {
case <-ctx.Done():
ticker.Stop()
return
case <-ticker.C:
s.processPendingBids()
}
}
ticker.Stop()
// tryProcessPendingBid validates and stores the bid once its proposer preferences have arrived,
// dropping bids whose slot is no longer current or whose preference matching fails.
func (s *executionPayloadBidService) tryProcessPendingBid(_ context.Context, key pendingBidKey, msg *cltypes.SignedExecutionPayloadBid) (func(), bool) {
// Check if bid slot is still valid
currentSlot := s.ethClock.GetCurrentSlot()
if key.slot != currentSlot && key.slot != currentSlot+1 {
log.Trace("Pending execution payload bid slot expired",
"slot", key.slot, "builderIndex", key.builderIndex)
return nil, true
}
}

// processPendingBids checks pending bids whose proposer preferences may have arrived.
func (s *executionPayloadBidService) processPendingBids() {
s.pendingBids.Range(func(key, value any) bool {
pendingKey := key.(pendingBidKey)
job := value.(*pendingBidJob)

// Check expiry
if time.Since(job.creationTime) > pendingBidExpiry {
s.pendingBids.Delete(pendingKey)
s.pendingCount.Add(-1)
log.Trace("Pending execution payload bid expired",
"slot", pendingKey.slot, "builderIndex", pendingKey.builderIndex)
return true
}

// Check if bid slot is still valid
currentSlot := s.ethClock.GetCurrentSlot()
if pendingKey.slot != currentSlot && pendingKey.slot != currentSlot+1 {
s.pendingBids.Delete(pendingKey)
s.pendingCount.Add(-1)
log.Trace("Pending execution payload bid slot expired",
"slot", pendingKey.slot, "builderIndex", pendingKey.builderIndex)
return true
}

preferences, ok, err := s.matchingProposerPreferences(job.msg)
if err != nil {
if errors.Is(err, errBidDependencyUnavailable) {
return true
}
s.pendingBids.Delete(pendingKey)
s.pendingCount.Add(-1)
log.Trace("Failed to match pending execution payload bid",
"slot", pendingKey.slot,
"builderIndex", pendingKey.builderIndex,
"err", err)
return true
}
if !ok {
return true // Preferences still not here, keep waiting
preferences, ok, err := s.matchingProposerPreferences(msg)
if err != nil {
if errors.Is(err, errBidDependencyUnavailable) {
return nil, false
}
log.Trace("Failed to match pending execution payload bid",
"slot", key.slot,
"builderIndex", key.builderIndex,
"err", err)
return nil, true
}
if !ok {
return nil, false // Preferences still not here, keep waiting
}

// Preferences arrived, remove from pending and process.
s.pendingBids.Delete(pendingKey)
s.pendingCount.Add(-1)

if err := s.validateAndStoreBid(job.msg, preferences); err != nil {
return func() {
if err := s.validateAndStoreBid(msg, preferences); err != nil {
log.Trace("Failed to process pending execution payload bid",
"slot", pendingKey.slot,
"builderIndex", pendingKey.builderIndex,
"slot", key.slot,
"builderIndex", key.builderIndex,
"err", err)
}
return true
})
}, true
}
Loading
Loading