From f76f7246da1d3f6e253f6d1988241f15638133ed Mon Sep 17 00:00:00 2001
From: yperbasis <andrey.ashikhmin@gmail.com>
Date: Thu, 2 Jul 2026 22:02:39 +0200
Subject: [PATCH 1/3] cl/phase1/network/services: pin execution payload bid
 pending-queue behavior

The bid service's pending-queue drop paths (expiry, stale slot) and the
background loop's enqueue-to-drain flow had no test coverage, unlike the
sibling execution payload and payload attestation services. Pin them
before consolidating the three token-identical queue implementations.

The loop test populates the forkchoice mock's plain maps before the
service constructor spawns the loop goroutine, so it is race-free; the
preferences arrive later via the thread-safe LRU pool.
---
 .../execution_payload_bid_service_test.go     | 91 +++++++++++++++++++
 1 file changed, 91 insertions(+)

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..a898192151f 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"
@@ -623,6 +624,96 @@ func TestExecutionPayloadBidServicePendingQueueCapConcurrent(t *testing.T) {
 	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.pendingBids.Store(key, &pendingBidJob{
+		msg:          msg,
+		creationTime: time.Now().Add(-pendingBidExpiry - time.Second), // expired
+	})
+	service.pendingCount.Store(1)
+
+	// Expiry is checked before the slot check, so no ethClock call is expected.
+	service.processPendingBids()
+
+	require.Equal(t, int32(0), service.pendingCount.Load())
+	_, exists := service.pendingBids.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.pendingBids.Store(key, &pendingBidJob{
+		msg:          msg,
+		creationTime: time.Now(),
+	})
+	service.pendingCount.Store(1)
+
+	// Bid slot 100 is neither current (200) nor next slot → dropped
+	ethClockMock.EXPECT().GetCurrentSlot().Return(uint64(200))
+
+	service.processPendingBids()
+
+	require.Equal(t, int32(0), service.pendingCount.Load())
+	_, exists := service.pendingBids.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.pendingCount.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.pendingCount.Load() == 0
+	}, 5*time.Second, 10*time.Millisecond)
+}
+
 func TestExecutionPayloadBidServiceDecodeGossipMessage(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()

From 2b14eaa5e587b89f24b87b698aaa36c26b1d3eed Mon Sep 17 00:00:00 2001
From: yperbasis <andrey.ashikhmin@gmail.com>
Date: Thu, 2 Jul 2026 22:09:29 +0200
Subject: [PATCH 2/3] cl/phase1/network/services: extract shared
 pendingJobQueue for Gloas gossip services

The execution payload, payload attestation, and execution payload bid
services each carried a token-identical pending-job queue (sync.Map +
atomic count + cond-signalled polling loop). Consolidate it into an
unexported generic pendingJobQueue[K, M]; readiness checks stay in
per-service tryProcess callbacks since they differ on purpose per topic
(block arrived / block arrived at current slot / preferences matched).

Per-service deltas, kept as queue parameters or callback logic:
- capacity: envelopes 1024, attestations 2048, bids 1024
- expiry: envelopes and attestations 30s, bids 12s (1 slot); tick 100ms everywhere
- keys: (blockRoot, envelopeHash) / (blockRoot, validatorIndex) / (builderIndex, slot)
- drop rules: attestations drop on stale slot; bids drop on stale
  slot and on non-dependency preference-matching errors, and keep
  waiting on errBidDependencyUnavailable
- the envelope service hashes the envelope to build its key and drops
  with a warning if hashing fails

The audit note that the bid service incremented pendingCount after
LoadOrStore is stale: that ordering existed at Gloas introduction
(f3ad70f1f7) but was fixed to increment-first in #21655 (a2f483ad07);
all three enqueues are token-identical today and the queue keeps that
increment-first accounting.

The tryProcess contract returns (process func(), done bool) so the queue
removes a finished job before running its processing, exactly as the
originals deleted before calling ProcessMessage/validateAndStoreBid;
a concurrent re-enqueue of the same key during processing is therefore
still stored rather than lost.

Deliberate normalizations, none affecting accounting or validation:
- envelope enqueue computes HashSSZ before the capacity check instead
  of after the capacity reservation; visible only when the queue is
  full (hash cost, and the hash-failure warning now logs even then)
- drop-path trace logs now precede the map removal instead of following
  it (same goroutine, not observable)
- the bid callback receives the loop context and ignores it, where
  processPendingBids previously took no context

The ticker-only schedulers in block, blob sidecar, and data column
sidecar services are intentionally not converted: they have no cond or
capacity accounting, use Store (replace) rather than LoadOrStore (drop)
on key collision, retry failed jobs until expiry instead of removing
them before processing, and the data column variant increments its
count after LoadOrStore and logs on capacity drops. Covering those
would need option knobs for enqueue collision policy, error retry, and
removal ordering, defeating the consolidation.
---
 .../services/execution_payload_bid_service.go | 158 +++++-------------
 .../execution_payload_bid_service_test.go     |  50 +++---
 .../services/execution_payload_service.go     | 121 +++-----------
 .../execution_payload_service_test.go         |  70 ++++----
 .../services/payload_attestation_service.go   | 129 +++-----------
 .../payload_attestation_service_test.go       |  62 +++----
 .../network/services/pending_job_queue.go     | 149 +++++++++++++++++
 7 files changed, 329 insertions(+), 410 deletions(-)
 create mode 100644 cl/phase1/network/services/pending_job_queue.go

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 a898192151f..b861ae1470b 100644
--- a/cl/phase1/network/services/execution_payload_bid_service_test.go
+++ b/cl/phase1/network/services/execution_payload_bid_service_test.go
@@ -56,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
 }
@@ -235,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
@@ -245,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)
 }
@@ -262,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
@@ -272,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)
 }
@@ -583,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)
 }
 
@@ -602,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++ {
@@ -615,9 +615,9 @@ 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
 	})
@@ -632,17 +632,17 @@ func TestExecutionPayloadBidServicePendingExpiry(t *testing.T) {
 
 	msg := newTestSignedExecutionPayloadBid(100, 1, 1000)
 	key := pendingBidKey{builderIndex: 1, slot: 100}
-	service.pendingBids.Store(key, &pendingBidJob{
+	service.pending.jobs.Store(key, &pendingJob[*cltypes.SignedExecutionPayloadBid]{
 		msg:          msg,
 		creationTime: time.Now().Add(-pendingBidExpiry - time.Second), // expired
 	})
-	service.pendingCount.Store(1)
+	service.pending.count.Store(1)
 
 	// Expiry is checked before the slot check, so no ethClock call is expected.
-	service.processPendingBids()
+	service.pending.processPending(context.Background())
 
-	require.Equal(t, int32(0), service.pendingCount.Load())
-	_, exists := service.pendingBids.Load(key)
+	require.Equal(t, int32(0), service.pending.count.Load())
+	_, exists := service.pending.jobs.Load(key)
 	require.False(t, exists)
 }
 
@@ -654,19 +654,19 @@ func TestExecutionPayloadBidServicePendingStaleSlotDropped(t *testing.T) {
 
 	msg := newTestSignedExecutionPayloadBid(100, 1, 1000)
 	key := pendingBidKey{builderIndex: 1, slot: 100}
-	service.pendingBids.Store(key, &pendingBidJob{
+	service.pending.jobs.Store(key, &pendingJob[*cltypes.SignedExecutionPayloadBid]{
 		msg:          msg,
 		creationTime: time.Now(),
 	})
-	service.pendingCount.Store(1)
+	service.pending.count.Store(1)
 
 	// Bid slot 100 is neither current (200) nor next slot → dropped
 	ethClockMock.EXPECT().GetCurrentSlot().Return(uint64(200))
 
-	service.processPendingBids()
+	service.pending.processPending(context.Background())
 
-	require.Equal(t, int32(0), service.pendingCount.Load())
-	_, exists := service.pendingBids.Load(key)
+	require.Equal(t, int32(0), service.pending.count.Load())
+	_, exists := service.pending.jobs.Load(key)
 	require.False(t, exists)
 }
 
@@ -703,14 +703,14 @@ func TestExecutionPayloadBidServiceLoopProcessesQueuedBid(t *testing.T) {
 
 	// No preferences yet → queued as pending
 	require.NoError(t, service.ProcessMessage(context.Background(), nil, msg))
-	require.Equal(t, int32(1), impl.pendingCount.Load())
+	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.pendingCount.Load() == 0
+		return found && impl.pending.count.Load() == 0
 	}, 5*time.Second, 10*time.Millisecond)
 }
 
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/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 <http://www.gnu.org/licenses/>.
+
+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
+	})
+}

From 174d4d33f96b52a48c01cbae5bf5ac1ff3f4af2b Mon Sep 17 00:00:00 2001
From: yperbasis <andrey.ashikhmin@gmail.com>
Date: Thu, 2 Jul 2026 22:11:31 +0200
Subject: [PATCH 3/3] cl/phase1/network/services: dedupe
 isLocalValidatorProposer into localProposerChecker

The aggregate-and-proof and sync-contribution services carried
token-identical isLocalValidatorProposer methods (docstring included)
along with duplicated proposerIndicesCache fields and LRU setup. Move
them to a localProposerChecker type embedded in both services, keeping
call sites unchanged via method promotion. Each service still owns its
own cache instance, as before.
---
 .../services/aggregate_and_proof_service.go   | 45 +----------
 .../services/local_proposer_checker.go        | 78 +++++++++++++++++++
 .../services/sync_contribution_service.go     | 46 +----------
 3 files changed, 83 insertions(+), 86 deletions(-)
 create mode 100644 cl/phase1/network/services/local_proposer_checker.go

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/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 <http://www.gnu.org/licenses/>.
+
+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/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).