diff --git a/cl/phase1/network/services/aggregate_and_proof_service.go b/cl/phase1/network/services/aggregate_and_proof_service.go
index 96dca162364..0f347190742 100644
--- a/cl/phase1/network/services/aggregate_and_proof_service.go
+++ b/cl/phase1/network/services/aggregate_and_proof_service.go
@@ -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
@@ -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,
@@ -112,7 +106,7 @@ func NewAggregateAndProofService(
batchSignatureVerifier: batchSignatureVerifier,
seenAggreatorIndexes: seenAggCache,
validatorParams: validatorParams,
- proposerIndicesCache: proposerIndicesCache,
+ localProposerChecker: newLocalProposerChecker(beaconCfg),
}
go a.loop(ctx)
return a
@@ -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()},
diff --git a/cl/phase1/network/services/execution_payload_bid_service.go b/cl/phase1/network/services/execution_payload_bid_service.go
index ff2b95565d5..ec60f3e79e9 100644
--- a/cl/phase1/network/services/execution_payload_bid_service.go
+++ b/cl/phase1/network/services/execution_payload_bid_service.go
@@ -20,8 +20,6 @@ import (
"context"
"errors"
"fmt"
- "sync"
- "sync/atomic"
"time"
"github.com/erigontech/erigon/cl/beacon/beaconevents"
@@ -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 (
@@ -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.
@@ -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}
}
@@ -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
}
diff --git a/cl/phase1/network/services/execution_payload_bid_service_test.go b/cl/phase1/network/services/execution_payload_bid_service_test.go
index ff79295c58c..b861ae1470b 100644
--- a/cl/phase1/network/services/execution_payload_bid_service_test.go
+++ b/cl/phase1/network/services/execution_payload_bid_service_test.go
@@ -6,6 +6,7 @@ import (
"fmt"
"sync"
"testing"
+ "time"
"github.com/stretchr/testify/require"
"go.uber.org/mock/gomock"
@@ -55,8 +56,8 @@ func setupExecutionPayloadBidService(t *testing.T, ctrl *gomock.Controller) (
epbsPool: epbsPool,
emitters: beaconevents.NewEventEmitter(),
seenCache: seenCache,
- pendingCond: sync.NewCond(&sync.Mutex{}),
}
+ service.pending = service.newPendingQueue()
return service, mockSyncedData, ethClockMock, fcMock, epbsPool
}
@@ -234,7 +235,7 @@ func TestExecutionPayloadBidServiceWaitsForMatchingDependentRootPreference(t *te
ethClockMock.EXPECT().GetCurrentSlot().Return(uint64(100))
require.NoError(t, service.ProcessMessage(context.Background(), nil, msg))
- require.Equal(t, int32(1), service.pendingCount.Load())
+ require.Equal(t, int32(1), service.pending.count.Load())
addPreferencesToPool(epbsPool, 100)
fcMock.ExecutionPayloadStatusMap[msg.Message.ParentBlockHash] = execution_client.PayloadStatusValidated
@@ -244,8 +245,8 @@ func TestExecutionPayloadBidServiceWaitsForMatchingDependentRootPreference(t *te
return nil
})
- service.processPendingBids()
- require.Equal(t, int32(0), service.pendingCount.Load())
+ service.pending.processPending(context.Background())
+ require.Equal(t, int32(0), service.pending.count.Load())
_, found := epbsPool.HighestBids.Get(pool.HighestBidKey{Slot: 100, ParentBlockHash: msg.Message.ParentBlockHash, ParentBlockRoot: msg.Message.ParentBlockRoot})
require.True(t, found)
}
@@ -261,7 +262,7 @@ func TestExecutionPayloadBidServiceWaitsForParentState(t *testing.T) {
ethClockMock.EXPECT().GetCurrentSlot().Return(uint64(100))
require.NoError(t, service.ProcessMessage(context.Background(), nil, msg))
- require.Equal(t, int32(1), service.pendingCount.Load())
+ require.Equal(t, int32(1), service.pending.count.Load())
fcMock.StateAtBlockRootVal[msg.Message.ParentBlockRoot] = newBidParentState(service.beaconCfg, testDependentRoot)
fcMock.ExecutionPayloadStatusMap[msg.Message.ParentBlockHash] = execution_client.PayloadStatusValidated
@@ -271,8 +272,8 @@ func TestExecutionPayloadBidServiceWaitsForParentState(t *testing.T) {
return nil
})
- service.processPendingBids()
- require.Equal(t, int32(0), service.pendingCount.Load())
+ service.pending.processPending(context.Background())
+ require.Equal(t, int32(0), service.pending.count.Load())
_, found := epbsPool.HighestBids.Get(pool.HighestBidKey{Slot: 100, ParentBlockHash: msg.Message.ParentBlockHash, ParentBlockRoot: msg.Message.ParentBlockRoot})
require.True(t, found)
}
@@ -582,16 +583,16 @@ func TestExecutionPayloadBidServicePendingQueueCap(t *testing.T) {
service, _, _, _, _ := setupExecutionPayloadBidService(t, ctrl)
// Fill the queue to the cap
- service.pendingCount.Store(maxPendingBids)
+ service.pending.count.Store(maxPendingBids)
msg := newTestSignedExecutionPayloadBid(100, 999, 1000)
service.queuePendingBid(msg)
// Should still be at cap — new item was rejected
- require.Equal(t, int32(maxPendingBids), service.pendingCount.Load())
+ require.Equal(t, int32(maxPendingBids), service.pending.count.Load())
key := pendingBidKey{builderIndex: 999, slot: 100}
- _, exists := service.pendingBids.Load(key)
+ _, exists := service.pending.jobs.Load(key)
require.False(t, exists)
}
@@ -601,7 +602,7 @@ func TestExecutionPayloadBidServicePendingQueueCapConcurrent(t *testing.T) {
service, _, _, _, _ := setupExecutionPayloadBidService(t, ctrl)
- service.pendingCount.Store(maxPendingBids - 5)
+ service.pending.count.Store(maxPendingBids - 5)
var wg sync.WaitGroup
for i := 0; i < 100; i++ {
@@ -614,15 +615,105 @@ func TestExecutionPayloadBidServicePendingQueueCapConcurrent(t *testing.T) {
}
wg.Wait()
- require.Equal(t, int32(maxPendingBids), service.pendingCount.Load())
+ require.Equal(t, int32(maxPendingBids), service.pending.count.Load())
stored := 0
- service.pendingBids.Range(func(_, _ any) bool {
+ service.pending.jobs.Range(func(_, _ any) bool {
stored++
return true
})
require.Equal(t, 5, stored)
}
+func TestExecutionPayloadBidServicePendingExpiry(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+
+ service, _, _, _, _ := setupExecutionPayloadBidService(t, ctrl)
+
+ msg := newTestSignedExecutionPayloadBid(100, 1, 1000)
+ key := pendingBidKey{builderIndex: 1, slot: 100}
+ service.pending.jobs.Store(key, &pendingJob[*cltypes.SignedExecutionPayloadBid]{
+ msg: msg,
+ creationTime: time.Now().Add(-pendingBidExpiry - time.Second), // expired
+ })
+ service.pending.count.Store(1)
+
+ // Expiry is checked before the slot check, so no ethClock call is expected.
+ service.pending.processPending(context.Background())
+
+ require.Equal(t, int32(0), service.pending.count.Load())
+ _, exists := service.pending.jobs.Load(key)
+ require.False(t, exists)
+}
+
+func TestExecutionPayloadBidServicePendingStaleSlotDropped(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+
+ service, _, ethClockMock, _, _ := setupExecutionPayloadBidService(t, ctrl)
+
+ msg := newTestSignedExecutionPayloadBid(100, 1, 1000)
+ key := pendingBidKey{builderIndex: 1, slot: 100}
+ service.pending.jobs.Store(key, &pendingJob[*cltypes.SignedExecutionPayloadBid]{
+ msg: msg,
+ creationTime: time.Now(),
+ })
+ service.pending.count.Store(1)
+
+ // Bid slot 100 is neither current (200) nor next slot → dropped
+ ethClockMock.EXPECT().GetCurrentSlot().Return(uint64(200))
+
+ service.pending.processPending(context.Background())
+
+ require.Equal(t, int32(0), service.pending.count.Load())
+ _, exists := service.pending.jobs.Load(key)
+ require.False(t, exists)
+}
+
+// TestExecutionPayloadBidServiceLoopProcessesQueuedBid exercises the background
+// loop end-to-end: a bid queued while proposer preferences are missing is
+// picked up and validated once the preferences arrive.
+func TestExecutionPayloadBidServiceLoopProcessesQueuedBid(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+
+ mockSyncedData := synced_data_mock.NewMockSyncedData(ctrl)
+ ethClockMock := eth_clock.NewMockEthereumClock(ctrl)
+ fcMock := forkchoice_mock.NewForkChoiceStorageMock(t)
+ epbsPool := pool.NewEpbsPool()
+ beaconCfg := clparams.MainnetBeaconConfig
+ beaconCfg.SlotsPerEpoch = 32
+ beaconCfg.SlotsPerHistoricalRoot = 8192
+ beaconCfg.MinSeedLookahead = 1
+ beaconCfg.DomainBeaconBuilder = [4]byte{0x0B, 0x00, 0x00, 0x00}
+
+ msg := newTestSignedExecutionPayloadBid(100, 1, 1000)
+ // Populate forkchoice mock before the service (and its loop goroutine) starts.
+ fcMock.StateAtBlockRootVal[msg.Message.ParentBlockRoot] = newBidParentState(&beaconCfg, testDependentRoot)
+ fcMock.ExecutionPayloadStatusMap[msg.Message.ParentBlockHash] = execution_client.PayloadStatusValidated
+ fcMock.Headers[msg.Message.ParentBlockRoot] = &cltypes.BeaconBlockHeader{}
+
+ ethClockMock.EXPECT().GetCurrentSlot().Return(uint64(100)).AnyTimes()
+ mockSyncedData.EXPECT().ViewHeadState(gomock.Any()).Return(nil).AnyTimes()
+
+ ctx, cancel := context.WithCancel(context.Background())
+ t.Cleanup(cancel)
+ service := NewExecutionPayloadBidService(ctx, mockSyncedData, fcMock, ethClockMock, &beaconCfg, epbsPool, beaconevents.NewEventEmitter())
+ impl := service.(*executionPayloadBidService)
+
+ // No preferences yet → queued as pending
+ require.NoError(t, service.ProcessMessage(context.Background(), nil, msg))
+ require.Equal(t, int32(1), impl.pending.count.Load())
+
+ addPreferencesToPool(epbsPool, 100)
+
+ bidKey := pool.HighestBidKey{Slot: 100, ParentBlockHash: msg.Message.ParentBlockHash, ParentBlockRoot: msg.Message.ParentBlockRoot}
+ require.Eventually(t, func() bool {
+ _, found := epbsPool.HighestBids.Get(bidKey)
+ return found && impl.pending.count.Load() == 0
+ }, 5*time.Second, 10*time.Millisecond)
+}
+
func TestExecutionPayloadBidServiceDecodeGossipMessage(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
diff --git a/cl/phase1/network/services/execution_payload_service.go b/cl/phase1/network/services/execution_payload_service.go
index cdac0bf5c17..9cb80fdbb7e 100644
--- a/cl/phase1/network/services/execution_payload_service.go
+++ b/cl/phase1/network/services/execution_payload_service.go
@@ -20,8 +20,6 @@ import (
"context"
"errors"
"fmt"
- "sync"
- "sync/atomic"
"time"
"github.com/erigontech/erigon/cl/beacon/beaconevents"
@@ -52,12 +50,6 @@ type pendingEnvelopeKey struct {
envelopeHash common.Hash
}
-// envelopeJob represents a pending envelope waiting for its block to arrive
-type envelopeJob struct {
- envelope *cltypes.SignedExecutionPayloadEnvelope
- creationTime time.Time
-}
-
const (
seenEnvelopeCacheSize = 1000
pendingEnvelopeExpiry = 30 * time.Second
@@ -74,9 +66,7 @@ type executionPayloadService struct {
seenEnvelopesCache *lru.Cache[seenEnvelopeKey, struct{}]
// Pending envelopes waiting for block to arrive
- pendingEnvelopes sync.Map // pendingEnvelopeKey -> *envelopeJob
- pendingCount atomic.Int32
- pendingCond *sync.Cond
+ pending *pendingJobQueue[pendingEnvelopeKey, *cltypes.SignedExecutionPayloadEnvelope]
}
// NewExecutionPayloadService creates a new execution payload service
@@ -95,12 +85,20 @@ func NewExecutionPayloadService(
beaconCfg: beaconCfg,
emitters: emitters,
seenEnvelopesCache: seenEnvelopesCache,
- pendingCond: sync.NewCond(&sync.Mutex{}),
}
- go s.loop(ctx)
+ s.pending = s.newPendingQueue()
+ go s.pending.loop(ctx)
return s
}
+func (s *executionPayloadService) newPendingQueue() *pendingJobQueue[pendingEnvelopeKey, *cltypes.SignedExecutionPayloadEnvelope] {
+ return newPendingJobQueue(maxPendingEnvelopes, pendingEnvelopeExpiry, pendingEnvelopeCheckInterval,
+ s.tryProcessPendingEnvelope,
+ func(key pendingEnvelopeKey) {
+ log.Trace("Pending envelope expired", "blockRoot", key.blockRoot)
+ })
+}
+
func (s *executionPayloadService) Names() []string {
return []string{gossip.TopicNameExecutionPayload}
}
@@ -200,104 +198,29 @@ func (s *executionPayloadService) ProcessMessage(ctx context.Context, _ *uint64,
// queuePendingEnvelope adds an envelope to the pending queue for later processing
func (s *executionPayloadService) queuePendingEnvelope(blockRoot common.Hash, envelope *cltypes.SignedExecutionPayloadEnvelope) {
- if s.pendingCount.Add(1) > maxPendingEnvelopes {
- s.pendingCount.Add(-1)
- return
- }
-
// Compute envelope hash to allow multiple candidates per block
envelopeHash, err := envelope.HashSSZ()
if err != nil {
- s.pendingCount.Add(-1)
log.Warn("Failed to hash envelope for pending queue", "blockRoot", blockRoot, "err", err)
return
}
- key := pendingEnvelopeKey{
+ s.pending.enqueue(pendingEnvelopeKey{
blockRoot: blockRoot,
envelopeHash: envelopeHash,
- }
-
- if _, loaded := s.pendingEnvelopes.LoadOrStore(key, &envelopeJob{
- envelope: envelope,
- creationTime: time.Now(),
- }); loaded {
- s.pendingCount.Add(-1)
- } else {
- s.pendingCond.L.Lock()
- s.pendingCond.Signal()
- s.pendingCond.L.Unlock()
- }
+ }, envelope)
}
-// loop is the background goroutine that processes pending envelopes
-func (s *executionPayloadService) 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 envelopes
- s.pendingCond.L.Lock()
- for s.pendingCount.Load() == 0 {
- // Check if context is cancelled
- select {
- case <-ctx.Done():
- s.pendingCond.L.Unlock()
- return
- default:
- }
- s.pendingCond.Wait()
- }
- s.pendingCond.L.Unlock()
-
- // Poll until all pending envelopes are processed
- ticker := time.NewTicker(pendingEnvelopeCheckInterval)
- for s.pendingCount.Load() > 0 {
- select {
- case <-ctx.Done():
- ticker.Stop()
- return
- case <-ticker.C:
- s.processPendingEnvelopes(ctx)
- }
- }
- ticker.Stop()
+// tryProcessPendingEnvelope re-runs full validation via ProcessMessage once the block has arrived.
+func (s *executionPayloadService) tryProcessPendingEnvelope(ctx context.Context, key pendingEnvelopeKey, envelope *cltypes.SignedExecutionPayloadEnvelope) (func(), bool) {
+ block, ok := s.forkchoiceStore.GetBlock(key.blockRoot)
+ if !ok || block == nil {
+ return nil, false // Block still not here, keep waiting
}
-}
-
-// processPendingEnvelopes checks and processes any pending envelopes whose blocks have arrived
-func (s *executionPayloadService) processPendingEnvelopes(ctx context.Context) {
- s.pendingEnvelopes.Range(func(key, value any) bool {
- pendingKey := key.(pendingEnvelopeKey)
- job := value.(*envelopeJob)
- // Check expiry
- if time.Since(job.creationTime) > pendingEnvelopeExpiry {
- s.pendingEnvelopes.Delete(pendingKey)
- s.pendingCount.Add(-1)
- log.Trace("Pending envelope expired", "blockRoot", pendingKey.blockRoot)
- return true
+ return func() {
+ if err := s.ProcessMessage(ctx, nil, envelope); err != nil {
+ log.Trace("Failed to process pending envelope", "blockRoot", key.blockRoot, "err", err)
}
-
- // Check if block has arrived
- block, ok := s.forkchoiceStore.GetBlock(pendingKey.blockRoot)
- if !ok || block == nil {
- return true // Block still not here, keep waiting
- }
-
- // Block arrived, remove from pending and process
- s.pendingEnvelopes.Delete(pendingKey)
- s.pendingCount.Add(-1)
-
- // Re-run full validation via ProcessMessage
- if err := s.ProcessMessage(ctx, nil, job.envelope); err != nil {
- log.Trace("Failed to process pending envelope", "blockRoot", pendingKey.blockRoot, "err", err)
- }
- return true
- })
+ }, true
}
diff --git a/cl/phase1/network/services/execution_payload_service_test.go b/cl/phase1/network/services/execution_payload_service_test.go
index 9e7fe5535fd..b6ec322f911 100644
--- a/cl/phase1/network/services/execution_payload_service_test.go
+++ b/cl/phase1/network/services/execution_payload_service_test.go
@@ -84,7 +84,7 @@ func TestExecutionPayloadServiceBlockNotFound(t *testing.T) {
// Verify envelope was queued (check internal state)
impl := service.(*executionPayloadService)
- require.Equal(t, int32(1), impl.pendingCount.Load())
+ require.Equal(t, int32(1), impl.pending.count.Load())
// Now add block to forkchoice
fcu.Blocks[blockRoot] = &cltypes.SignedBeaconBlock{
@@ -206,13 +206,13 @@ func TestExecutionPayloadServicePendingEnvelopeExpiry(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
- // Create service directly to access internals
+ // Create service directly to access internals; the background loop is not started
impl := &executionPayloadService{
forkchoiceStore: forkchoiceMock,
beaconCfg: cfg,
emitters: beaconevents.NewEventEmitter(),
- pendingCond: nil, // Don't start background loop
}
+ impl.pending = impl.newPendingQueue()
seenCache, err := lru.New[seenEnvelopeKey, struct{}]("seen_envelopes", seenEnvelopeCacheSize)
require.NoError(t, err)
impl.seenEnvelopesCache = seenCache
@@ -227,17 +227,17 @@ func TestExecutionPayloadServicePendingEnvelopeExpiry(t *testing.T) {
blockRoot: blockRoot,
envelopeHash: envelopeHash,
}
- impl.pendingEnvelopes.Store(key, &envelopeJob{
- envelope: envelope,
+ impl.pending.jobs.Store(key, &pendingJob[*cltypes.SignedExecutionPayloadEnvelope]{
+ msg: envelope,
creationTime: time.Now().Add(-pendingEnvelopeExpiry - time.Second), // expired
})
- impl.pendingCount.Store(1)
+ impl.pending.count.Store(1)
// Process pending - should remove expired
- impl.processPendingEnvelopes(ctx)
+ impl.pending.processPending(ctx)
- require.Equal(t, int32(0), impl.pendingCount.Load())
- _, exists := impl.pendingEnvelopes.Load(key)
+ require.Equal(t, int32(0), impl.pending.count.Load())
+ _, exists := impl.pending.jobs.Load(key)
require.False(t, exists)
}
@@ -247,13 +247,13 @@ func TestExecutionPayloadServicePendingEnvelopeProcessing(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
- // Create service directly to access internals
+ // Create service directly to access internals; the background loop is not started
impl := &executionPayloadService{
forkchoiceStore: forkchoiceMock,
beaconCfg: cfg,
emitters: beaconevents.NewEventEmitter(),
- pendingCond: nil,
}
+ impl.pending = impl.newPendingQueue()
seenCache, err := lru.New[seenEnvelopeKey, struct{}]("seen_envelopes", seenEnvelopeCacheSize)
require.NoError(t, err)
impl.seenEnvelopesCache = seenCache
@@ -268,15 +268,15 @@ func TestExecutionPayloadServicePendingEnvelopeProcessing(t *testing.T) {
blockRoot: blockRoot,
envelopeHash: envelopeHash,
}
- impl.pendingEnvelopes.Store(key, &envelopeJob{
- envelope: envelope,
+ impl.pending.jobs.Store(key, &pendingJob[*cltypes.SignedExecutionPayloadEnvelope]{
+ msg: envelope,
creationTime: time.Now(),
})
- impl.pendingCount.Store(1)
+ impl.pending.count.Store(1)
// Block not yet available - should keep pending
- impl.processPendingEnvelopes(ctx)
- require.Equal(t, int32(1), impl.pendingCount.Load())
+ impl.pending.processPending(ctx)
+ require.Equal(t, int32(1), impl.pending.count.Load())
// Now add block
forkchoiceMock.Blocks[blockRoot] = &cltypes.SignedBeaconBlock{
@@ -286,9 +286,9 @@ func TestExecutionPayloadServicePendingEnvelopeProcessing(t *testing.T) {
}
// Process again - should process and remove
- impl.processPendingEnvelopes(ctx)
- require.Equal(t, int32(0), impl.pendingCount.Load())
- _, exists := impl.pendingEnvelopes.Load(key)
+ impl.pending.processPending(ctx)
+ require.Equal(t, int32(0), impl.pending.count.Load())
+ _, exists := impl.pending.jobs.Load(key)
require.False(t, exists)
// Envelope should be marked as seen
@@ -305,8 +305,8 @@ func TestExecutionPayloadServiceMultiplePendingForSameBlock(t *testing.T) {
forkchoiceStore: forkchoiceMock,
beaconCfg: cfg,
emitters: beaconevents.NewEventEmitter(),
- pendingCond: nil,
}
+ impl.pending = impl.newPendingQueue()
seenCache, err := lru.New[seenEnvelopeKey, struct{}]("seen_envelopes", seenEnvelopeCacheSize)
require.NoError(t, err)
impl.seenEnvelopesCache = seenCache
@@ -321,15 +321,15 @@ func TestExecutionPayloadServiceMultiplePendingForSameBlock(t *testing.T) {
hash2, _ := envelope2.HashSSZ()
// Add both as pending
- impl.pendingEnvelopes.Store(pendingEnvelopeKey{blockRoot, hash1}, &envelopeJob{
- envelope: envelope1,
+ impl.pending.jobs.Store(pendingEnvelopeKey{blockRoot, hash1}, &pendingJob[*cltypes.SignedExecutionPayloadEnvelope]{
+ msg: envelope1,
creationTime: time.Now(),
})
- impl.pendingEnvelopes.Store(pendingEnvelopeKey{blockRoot, hash2}, &envelopeJob{
- envelope: envelope2,
+ impl.pending.jobs.Store(pendingEnvelopeKey{blockRoot, hash2}, &pendingJob[*cltypes.SignedExecutionPayloadEnvelope]{
+ msg: envelope2,
creationTime: time.Now(),
})
- impl.pendingCount.Store(2)
+ impl.pending.count.Store(2)
// Add block
forkchoiceMock.Blocks[blockRoot] = &cltypes.SignedBeaconBlock{
@@ -339,9 +339,9 @@ func TestExecutionPayloadServiceMultiplePendingForSameBlock(t *testing.T) {
}
// Process - both should be processed
- impl.processPendingEnvelopes(ctx)
+ impl.pending.processPending(ctx)
- require.Equal(t, int32(0), impl.pendingCount.Load())
+ require.Equal(t, int32(0), impl.pending.count.Load())
require.True(t, impl.seenEnvelopesCache.Contains(seenEnvelopeKey{blockRoot, 1}))
require.True(t, impl.seenEnvelopesCache.Contains(seenEnvelopeKey{blockRoot, 2}))
}
@@ -357,20 +357,20 @@ func TestExecutionPayloadServicePendingQueueCap(t *testing.T) {
beaconCfg: cfg,
emitters: beaconevents.NewEventEmitter(),
seenEnvelopesCache: seenCache,
- pendingCond: sync.NewCond(&sync.Mutex{}),
}
+ impl.pending = impl.newPendingQueue()
- impl.pendingCount.Store(maxPendingEnvelopes)
+ impl.pending.count.Store(maxPendingEnvelopes)
blockRoot := common.HexToHash("0xffff")
envelope := newTestSignedEnvelope(100, blockRoot, 999)
impl.queuePendingEnvelope(blockRoot, envelope)
- require.Equal(t, int32(maxPendingEnvelopes), impl.pendingCount.Load())
+ require.Equal(t, int32(maxPendingEnvelopes), impl.pending.count.Load())
envelopeHash, err := envelope.HashSSZ()
require.NoError(t, err)
- _, exists := impl.pendingEnvelopes.Load(pendingEnvelopeKey{blockRoot, envelopeHash})
+ _, exists := impl.pending.jobs.Load(pendingEnvelopeKey{blockRoot, envelopeHash})
require.False(t, exists)
}
@@ -385,10 +385,10 @@ func TestExecutionPayloadServicePendingQueueCapConcurrent(t *testing.T) {
beaconCfg: cfg,
emitters: beaconevents.NewEventEmitter(),
seenEnvelopesCache: seenCache,
- pendingCond: sync.NewCond(&sync.Mutex{}),
}
+ impl.pending = impl.newPendingQueue()
- impl.pendingCount.Store(maxPendingEnvelopes - 5)
+ impl.pending.count.Store(maxPendingEnvelopes - 5)
var wg sync.WaitGroup
for i := 0; i < 100; i++ {
@@ -402,9 +402,9 @@ func TestExecutionPayloadServicePendingQueueCapConcurrent(t *testing.T) {
}
wg.Wait()
- require.Equal(t, int32(maxPendingEnvelopes), impl.pendingCount.Load())
+ require.Equal(t, int32(maxPendingEnvelopes), impl.pending.count.Load())
stored := 0
- impl.pendingEnvelopes.Range(func(_, _ any) bool {
+ impl.pending.jobs.Range(func(_, _ any) bool {
stored++
return true
})
diff --git a/cl/phase1/network/services/local_proposer_checker.go b/cl/phase1/network/services/local_proposer_checker.go
new file mode 100644
index 00000000000..e4d59940ee7
--- /dev/null
+++ b/cl/phase1/network/services/local_proposer_checker.go
@@ -0,0 +1,78 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see .
+
+package services
+
+import (
+ "slices"
+
+ "github.com/erigontech/erigon/cl/clparams"
+ "github.com/erigontech/erigon/cl/phase1/core/state"
+ "github.com/erigontech/erigon/cl/phase1/core/state/lru"
+)
+
+type localProposerChecker struct {
+ beaconCfg *clparams.BeaconChainConfig
+
+ // Cached proposer indices per epoch (for current epoch check)
+ proposerIndicesCache *lru.Cache[uint64, []uint64]
+}
+
+func newLocalProposerChecker(beaconCfg *clparams.BeaconChainConfig) localProposerChecker {
+ proposerIndicesCache, err := lru.New[uint64, []uint64]("proposerIndices", 3)
+ if err != nil {
+ panic(err)
+ }
+ return localProposerChecker{
+ beaconCfg: beaconCfg,
+ proposerIndicesCache: proposerIndicesCache,
+ }
+}
+
+// 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 (c localProposerChecker) 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 := c.proposerIndicesCache.Get(currentEpoch)
+ if !ok {
+ var err error
+ currentProposers, err = headState.GetBeaconProposerIndices(currentEpoch)
+ if err == nil {
+ c.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(c.beaconCfg.SlotsPerEpoch)
+ for i := slotsPerEpoch; i < lookahead.Length(); i++ {
+ proposerIndex := lookahead.Get(i)
+ if slices.Contains(localValidators, proposerIndex) {
+ return true
+ }
+ }
+ return false
+}
diff --git a/cl/phase1/network/services/payload_attestation_service.go b/cl/phase1/network/services/payload_attestation_service.go
index 7484f9b8d5e..b0342f7849d 100644
--- a/cl/phase1/network/services/payload_attestation_service.go
+++ b/cl/phase1/network/services/payload_attestation_service.go
@@ -20,8 +20,6 @@ import (
"context"
"errors"
"fmt"
- "sync"
- "sync/atomic"
"time"
"github.com/erigontech/erigon/cl/beacon/beaconevents"
@@ -49,12 +47,6 @@ type pendingPayloadAttestationKey struct {
validatorIndex uint64
}
-// pendingPayloadAttestationJob represents a pending attestation waiting for its block.
-type pendingPayloadAttestationJob struct {
- msg *cltypes.PayloadAttestationMessage
- creationTime time.Time
-}
-
const (
// seenPayloadAttestationCacheSize: PTC has 512 validators per slot.
// With clock disparity, we may see attestations for ~2 slots.
@@ -75,9 +67,7 @@ type payloadAttestationService struct {
seenAttestationsCache *lru.Cache[seenPayloadAttestationKey, struct{}]
// Pending attestations waiting for block to arrive
- pendingAttestations sync.Map // pendingPayloadAttestationKey -> *pendingPayloadAttestationJob
- pendingCount atomic.Int32
- pendingCond *sync.Cond
+ pending *pendingJobQueue[pendingPayloadAttestationKey, *cltypes.PayloadAttestationMessage]
}
// NewPayloadAttestationService creates a new payload attestation service.
@@ -99,12 +89,20 @@ func NewPayloadAttestationService(
netCfg: netCfg,
emitters: emitters,
seenAttestationsCache: seenCache,
- pendingCond: sync.NewCond(&sync.Mutex{}),
}
- go s.loop(ctx)
+ s.pending = s.newPendingQueue()
+ go s.pending.loop(ctx)
return s
}
+func (s *payloadAttestationService) newPendingQueue() *pendingJobQueue[pendingPayloadAttestationKey, *cltypes.PayloadAttestationMessage] {
+ return newPendingJobQueue(maxPendingAttestations, pendingPayloadAttestationExpiry, pendingPayloadAttestationCheckInterval,
+ s.tryProcessPendingAttestation,
+ func(key pendingPayloadAttestationKey) {
+ log.Trace("Pending payload attestation expired", "blockRoot", key.blockRoot)
+ })
+}
+
func (s *payloadAttestationService) Names() []string {
return []string{gossip.TopicNamePayloadAttestation}
}
@@ -196,102 +194,27 @@ func (s *payloadAttestationService) ProcessMessage(ctx context.Context, _ *uint6
// queuePendingAttestation adds an attestation to the pending queue for later processing.
func (s *payloadAttestationService) queuePendingAttestation(blockRoot common.Hash, msg *cltypes.PayloadAttestationMessage) {
- if s.pendingCount.Add(1) > maxPendingAttestations {
- s.pendingCount.Add(-1)
- return
- }
-
- key := pendingPayloadAttestationKey{
+ s.pending.enqueue(pendingPayloadAttestationKey{
blockRoot: blockRoot,
validatorIndex: msg.ValidatorIndex,
- }
-
- if _, loaded := s.pendingAttestations.LoadOrStore(key, &pendingPayloadAttestationJob{
- 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 attestations.
-func (s *payloadAttestationService) 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 attestations
- 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 attestations are processed
- ticker := time.NewTicker(pendingPayloadAttestationCheckInterval)
- for s.pendingCount.Load() > 0 {
- select {
- case <-ctx.Done():
- ticker.Stop()
- return
- case <-ticker.C:
- s.processPendingAttestations(ctx)
- }
- }
- ticker.Stop()
+// tryProcessPendingAttestation re-runs validation via ProcessMessage once the block has arrived,
+// dropping attestations that are no longer for the current slot.
+func (s *payloadAttestationService) tryProcessPendingAttestation(ctx context.Context, key pendingPayloadAttestationKey, msg *cltypes.PayloadAttestationMessage) (func(), bool) {
+ if !s.ethClock.IsSlotCurrentSlotWithMaximumClockDisparity(msg.Data.Slot) {
+ log.Trace("Pending payload attestation slot mismatch", "blockRoot", key.blockRoot)
+ return nil, true
}
-}
-
-// processPendingAttestations checks and processes any pending attestations whose blocks have arrived.
-func (s *payloadAttestationService) processPendingAttestations(ctx context.Context) {
- s.pendingAttestations.Range(func(key, value any) bool {
- pendingKey := key.(pendingPayloadAttestationKey)
- job := value.(*pendingPayloadAttestationJob)
-
- // Check expiry
- if time.Since(job.creationTime) > pendingPayloadAttestationExpiry {
- s.pendingAttestations.Delete(pendingKey)
- s.pendingCount.Add(-1)
- log.Trace("Pending payload attestation expired", "blockRoot", pendingKey.blockRoot)
- return true
- }
- // Check if attestation is still for current slot (with clock disparity allowance)
- if !s.ethClock.IsSlotCurrentSlotWithMaximumClockDisparity(job.msg.Data.Slot) {
- s.pendingAttestations.Delete(pendingKey)
- s.pendingCount.Add(-1)
- log.Trace("Pending payload attestation slot mismatch", "blockRoot", pendingKey.blockRoot)
- return true
- }
-
- // Check if block has arrived
- if _, ok := s.forkchoiceStore.GetHeader(pendingKey.blockRoot); !ok {
- return true // Block still not here, keep waiting
- }
-
- // Block arrived, remove from pending and process
- s.pendingAttestations.Delete(pendingKey)
- s.pendingCount.Add(-1)
+ if _, ok := s.forkchoiceStore.GetHeader(key.blockRoot); !ok {
+ return nil, false // Block still not here, keep waiting
+ }
- // Re-run validation via ProcessMessage
- if err := s.ProcessMessage(ctx, nil, job.msg); err != nil {
- log.Trace("Failed to process pending payload attestation", "blockRoot", pendingKey.blockRoot, "err", err)
+ return func() {
+ if err := s.ProcessMessage(ctx, nil, msg); err != nil {
+ log.Trace("Failed to process pending payload attestation", "blockRoot", key.blockRoot, "err", err)
}
- return true
- })
+ }, true
}
diff --git a/cl/phase1/network/services/payload_attestation_service_test.go b/cl/phase1/network/services/payload_attestation_service_test.go
index 3ef66da5df3..f4f2acc70b3 100644
--- a/cl/phase1/network/services/payload_attestation_service_test.go
+++ b/cl/phase1/network/services/payload_attestation_service_test.go
@@ -48,8 +48,8 @@ func setupPayloadAttestationService(t *testing.T, ctrl *gomock.Controller) (*pay
netCfg: nil, // Not used in current implementation
seenAttestationsCache: seenCache,
emitters: beaconevents.NewEventEmitter(),
- pendingCond: sync.NewCond(&sync.Mutex{}), // Needed for queuePendingAttestation
}
+ service.pending = service.newPendingQueue()
return service, forkchoiceMock, ethClockMock
}
@@ -150,14 +150,14 @@ func TestPayloadAttestationServiceBlockNotFound(t *testing.T) {
require.NoError(t, err)
// Verify attestation was queued
- require.Equal(t, int32(1), service.pendingCount.Load())
+ require.Equal(t, int32(1), service.pending.count.Load())
// Verify the pending key
key := pendingPayloadAttestationKey{
blockRoot: blockRoot,
validatorIndex: 1,
}
- _, exists := service.pendingAttestations.Load(key)
+ _, exists := service.pending.jobs.Load(key)
require.True(t, exists)
}
@@ -255,17 +255,17 @@ func TestPayloadAttestationServicePendingExpiry(t *testing.T) {
blockRoot: blockRoot,
validatorIndex: 1,
}
- service.pendingAttestations.Store(key, &pendingPayloadAttestationJob{
+ service.pending.jobs.Store(key, &pendingJob[*cltypes.PayloadAttestationMessage]{
msg: msg,
creationTime: time.Now().Add(-pendingPayloadAttestationExpiry - time.Second), // expired
})
- service.pendingCount.Store(1)
+ service.pending.count.Store(1)
// Process pending - should remove expired
- service.processPendingAttestations(context.Background())
+ service.pending.processPending(context.Background())
- require.Equal(t, int32(0), service.pendingCount.Load())
- _, exists := service.pendingAttestations.Load(key)
+ require.Equal(t, int32(0), service.pending.count.Load())
+ _, exists := service.pending.jobs.Load(key)
require.False(t, exists)
}
@@ -283,20 +283,20 @@ func TestPayloadAttestationServicePendingSlotMismatch(t *testing.T) {
blockRoot: blockRoot,
validatorIndex: 1,
}
- service.pendingAttestations.Store(key, &pendingPayloadAttestationJob{
+ service.pending.jobs.Store(key, &pendingJob[*cltypes.PayloadAttestationMessage]{
msg: msg,
creationTime: time.Now(),
})
- service.pendingCount.Store(1)
+ service.pending.count.Store(1)
// Mock: slot 100 is no longer current
ethClockMock.EXPECT().IsSlotCurrentSlotWithMaximumClockDisparity(uint64(100)).Return(false)
// Process pending - should remove due to slot mismatch
- service.processPendingAttestations(context.Background())
+ service.pending.processPending(context.Background())
- require.Equal(t, int32(0), service.pendingCount.Load())
- _, exists := service.pendingAttestations.Load(key)
+ require.Equal(t, int32(0), service.pending.count.Load())
+ _, exists := service.pending.jobs.Load(key)
require.False(t, exists)
}
@@ -314,16 +314,16 @@ func TestPayloadAttestationServicePendingProcessing(t *testing.T) {
blockRoot: blockRoot,
validatorIndex: 42,
}
- service.pendingAttestations.Store(key, &pendingPayloadAttestationJob{
+ service.pending.jobs.Store(key, &pendingJob[*cltypes.PayloadAttestationMessage]{
msg: msg,
creationTime: time.Now(),
})
- service.pendingCount.Store(1)
+ service.pending.count.Store(1)
// First process: slot ok, but block not available
ethClockMock.EXPECT().IsSlotCurrentSlotWithMaximumClockDisparity(uint64(100)).Return(true)
- service.processPendingAttestations(context.Background())
- require.Equal(t, int32(1), service.pendingCount.Load()) // Still pending
+ service.pending.processPending(context.Background())
+ require.Equal(t, int32(1), service.pending.count.Load()) // Still pending
// Now add block header
fcu.Headers[blockRoot] = &cltypes.BeaconBlockHeader{
@@ -333,10 +333,10 @@ func TestPayloadAttestationServicePendingProcessing(t *testing.T) {
// Second process: slot ok, block available -> should process
// ProcessMessage will be called, which calls IsSlotCurrentSlotWithMaximumClockDisparity again
ethClockMock.EXPECT().IsSlotCurrentSlotWithMaximumClockDisparity(uint64(100)).Return(true).Times(2)
- service.processPendingAttestations(context.Background())
+ service.pending.processPending(context.Background())
- require.Equal(t, int32(0), service.pendingCount.Load())
- _, exists := service.pendingAttestations.Load(key)
+ require.Equal(t, int32(0), service.pending.count.Load())
+ _, exists := service.pending.jobs.Load(key)
require.False(t, exists)
// Attestation should be marked as seen
@@ -356,15 +356,15 @@ func TestPayloadAttestationServiceMultiplePendingForSameBlock(t *testing.T) {
msg2 := newTestPayloadAttestationMessage(100, 2, blockRoot)
// Add both as pending
- service.pendingAttestations.Store(pendingPayloadAttestationKey{blockRoot, 1}, &pendingPayloadAttestationJob{
+ service.pending.jobs.Store(pendingPayloadAttestationKey{blockRoot, 1}, &pendingJob[*cltypes.PayloadAttestationMessage]{
msg: msg1,
creationTime: time.Now(),
})
- service.pendingAttestations.Store(pendingPayloadAttestationKey{blockRoot, 2}, &pendingPayloadAttestationJob{
+ service.pending.jobs.Store(pendingPayloadAttestationKey{blockRoot, 2}, &pendingJob[*cltypes.PayloadAttestationMessage]{
msg: msg2,
creationTime: time.Now(),
})
- service.pendingCount.Store(2)
+ service.pending.count.Store(2)
// Add block header
fcu.Headers[blockRoot] = &cltypes.BeaconBlockHeader{
@@ -375,9 +375,9 @@ func TestPayloadAttestationServiceMultiplePendingForSameBlock(t *testing.T) {
ethClockMock.EXPECT().IsSlotCurrentSlotWithMaximumClockDisparity(uint64(100)).Return(true).Times(4)
// Process - both should be processed
- service.processPendingAttestations(context.Background())
+ service.pending.processPending(context.Background())
- require.Equal(t, int32(0), service.pendingCount.Load())
+ require.Equal(t, int32(0), service.pending.count.Load())
require.True(t, service.seenAttestationsCache.Contains(seenPayloadAttestationKey{100, 1}))
require.True(t, service.seenAttestationsCache.Contains(seenPayloadAttestationKey{100, 2}))
}
@@ -389,7 +389,7 @@ func TestPayloadAttestationServicePendingQueueCap(t *testing.T) {
service, _, _ := setupPayloadAttestationService(t, ctrl)
// Fill the queue to the cap
- service.pendingCount.Store(maxPendingAttestations)
+ service.pending.count.Store(maxPendingAttestations)
blockRoot := common.HexToHash("0xffff")
msg := newTestPayloadAttestationMessage(100, 999, blockRoot)
@@ -397,9 +397,9 @@ func TestPayloadAttestationServicePendingQueueCap(t *testing.T) {
service.queuePendingAttestation(blockRoot, msg)
// Should still be at cap — new item was rejected
- require.Equal(t, int32(maxPendingAttestations), service.pendingCount.Load())
+ require.Equal(t, int32(maxPendingAttestations), service.pending.count.Load())
key := pendingPayloadAttestationKey{blockRoot: blockRoot, validatorIndex: 999}
- _, exists := service.pendingAttestations.Load(key)
+ _, exists := service.pending.jobs.Load(key)
require.False(t, exists)
}
@@ -410,7 +410,7 @@ func TestPayloadAttestationServicePendingQueueCapConcurrent(t *testing.T) {
service, _, _ := setupPayloadAttestationService(t, ctrl)
// Start near cap so only a few slots remain
- service.pendingCount.Store(maxPendingAttestations - 5)
+ service.pending.count.Store(maxPendingAttestations - 5)
var wg sync.WaitGroup
for i := 0; i < 100; i++ {
@@ -424,9 +424,9 @@ func TestPayloadAttestationServicePendingQueueCapConcurrent(t *testing.T) {
}
wg.Wait()
- require.Equal(t, int32(maxPendingAttestations), service.pendingCount.Load())
+ require.Equal(t, int32(maxPendingAttestations), service.pending.count.Load())
stored := 0
- service.pendingAttestations.Range(func(_, _ any) bool {
+ service.pending.jobs.Range(func(_, _ any) bool {
stored++
return true
})
diff --git a/cl/phase1/network/services/pending_job_queue.go b/cl/phase1/network/services/pending_job_queue.go
new file mode 100644
index 00000000000..70bcef581b3
--- /dev/null
+++ b/cl/phase1/network/services/pending_job_queue.go
@@ -0,0 +1,149 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see .
+
+package services
+
+import (
+ "context"
+ "sync"
+ "sync/atomic"
+ "time"
+)
+
+type pendingJob[M any] struct {
+ msg M
+ creationTime time.Time
+}
+
+// pendingJobQueue holds gossip messages whose processing dependencies have not
+// arrived yet and retries them periodically until processed or expired.
+type pendingJobQueue[K comparable, M any] struct {
+ capacity int32
+ expiry time.Duration
+ tick time.Duration
+ // tryProcess decides a job's fate: done=false keeps it queued for the next
+ // tick; done=true removes it. A non-nil process func runs after the removal,
+ // so that a concurrent re-enqueue under the same key is not lost.
+ tryProcess func(ctx context.Context, key K, msg M) (process func(), done bool)
+ onExpired func(key K)
+
+ jobs sync.Map // K -> *pendingJob[M]
+ count atomic.Int32
+ cond *sync.Cond
+}
+
+func newPendingJobQueue[K comparable, M any](
+ capacity int32,
+ expiry time.Duration,
+ tick time.Duration,
+ tryProcess func(ctx context.Context, key K, msg M) (func(), bool),
+ onExpired func(key K),
+) *pendingJobQueue[K, M] {
+ return &pendingJobQueue[K, M]{
+ capacity: capacity,
+ expiry: expiry,
+ tick: tick,
+ tryProcess: tryProcess,
+ onExpired: onExpired,
+ cond: sync.NewCond(&sync.Mutex{}),
+ }
+}
+
+// enqueue adds a job unless the queue is at capacity or the key is already queued.
+func (q *pendingJobQueue[K, M]) enqueue(key K, msg M) {
+ if q.count.Add(1) > q.capacity {
+ q.count.Add(-1)
+ return
+ }
+
+ if _, loaded := q.jobs.LoadOrStore(key, &pendingJob[M]{
+ msg: msg,
+ creationTime: time.Now(),
+ }); loaded {
+ q.count.Add(-1)
+ } else {
+ q.cond.L.Lock()
+ q.cond.Signal()
+ q.cond.L.Unlock()
+ }
+}
+
+func (q *pendingJobQueue[K, M]) remove(key K) {
+ q.jobs.Delete(key)
+ q.count.Add(-1)
+}
+
+// loop is the background goroutine that retries pending jobs.
+func (q *pendingJobQueue[K, M]) loop(ctx context.Context) {
+ // Wake any blocked Wait() on context cancellation to prevent deadlock.
+ go func() {
+ <-ctx.Done()
+ q.cond.L.Lock()
+ q.cond.Broadcast()
+ q.cond.L.Unlock()
+ }()
+
+ for {
+ // Wait until there are pending jobs
+ q.cond.L.Lock()
+ for q.count.Load() == 0 {
+ select {
+ case <-ctx.Done():
+ q.cond.L.Unlock()
+ return
+ default:
+ }
+ q.cond.Wait()
+ }
+ q.cond.L.Unlock()
+
+ // Poll until all pending jobs are processed
+ ticker := time.NewTicker(q.tick)
+ for q.count.Load() > 0 {
+ select {
+ case <-ctx.Done():
+ ticker.Stop()
+ return
+ case <-ticker.C:
+ q.processPending(ctx)
+ }
+ }
+ ticker.Stop()
+ }
+}
+
+func (q *pendingJobQueue[K, M]) processPending(ctx context.Context) {
+ q.jobs.Range(func(key, value any) bool {
+ k := key.(K)
+ job := value.(*pendingJob[M])
+
+ if time.Since(job.creationTime) > q.expiry {
+ q.remove(k)
+ q.onExpired(k)
+ return true
+ }
+
+ process, done := q.tryProcess(ctx, k, job.msg)
+ if !done {
+ return true
+ }
+ q.remove(k)
+ if process != nil {
+ process()
+ }
+ return true
+ })
+}
diff --git a/cl/phase1/network/services/sync_contribution_service.go b/cl/phase1/network/services/sync_contribution_service.go
index 705486b9b72..b4b19497ab7 100644
--- a/cl/phase1/network/services/sync_contribution_service.go
+++ b/cl/phase1/network/services/sync_contribution_service.go
@@ -32,7 +32,6 @@ import (
"github.com/erigontech/erigon/cl/fork"
"github.com/erigontech/erigon/cl/gossip"
"github.com/erigontech/erigon/cl/phase1/core/state"
- "github.com/erigontech/erigon/cl/phase1/core/state/lru"
"github.com/erigontech/erigon/cl/utils"
"github.com/erigontech/erigon/cl/utils/bls"
"github.com/erigontech/erigon/cl/utils/eth_clock"
@@ -58,8 +57,8 @@ type syncContributionService struct {
ethClock eth_clock.EthereumClock
batchSignatureVerifier *BatchSignatureVerifier
validatorParams *validator_params.ValidatorParams
- proposerIndicesCache *lru.Cache[uint64, []uint64]
- test bool
+ localProposerChecker
+ test bool
mu sync.Mutex
}
@@ -82,10 +81,6 @@ func NewSyncContributionService(
validatorParams *validator_params.ValidatorParams,
test bool,
) SyncContributionService {
- proposerIndicesCache, err := lru.New[uint64, []uint64]("proposerIndices", 3)
- if err != nil {
- panic(err)
- }
return &syncContributionService{
syncedDataManager: syncedDataManager,
beaconCfg: beaconCfg,
@@ -95,7 +90,7 @@ func NewSyncContributionService(
emitters: emitters,
batchSignatureVerifier: batchSignatureVerifier,
validatorParams: validatorParams,
- proposerIndicesCache: proposerIndicesCache,
+ localProposerChecker: newLocalProposerChecker(beaconCfg),
test: test,
}
}
@@ -108,41 +103,6 @@ func (s *syncContributionService) IsMyGossipMessage(name string) bool {
return name == gossip.TopicNameSyncCommitteeContributionAndProof
}
-// 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 (s *syncContributionService) 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 := s.proposerIndicesCache.Get(currentEpoch)
- if !ok {
- var err error
- currentProposers, err = headState.GetBeaconProposerIndices(currentEpoch)
- if err == nil {
- s.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(s.beaconCfg.SlotsPerEpoch)
- for i := slotsPerEpoch; i < lookahead.Length(); i++ {
- proposerIndex := lookahead.Get(i)
- if slices.Contains(localValidators, proposerIndex) {
- return true
- }
- }
- return false
-}
-
func (s *syncContributionService) DecodeGossipMessage(pid peer.ID, data []byte, version clparams.StateVersion) (*SignedContributionAndProofForGossip, error) {
// Pre-allocate with config-aware aggregation bits size so DecodeSSZ uses the
// correct byte length for the current preset (minimal: 1 byte, mainnet: 16 bytes).