remove redundant justified checkpoint update in pullTips

**What type of PR is this?**
Other

**What does this PR do? Why is it needed?**
When an (potentially aggregated) attestation is received **before** the
block being voted for, Prysm queues this attestation, then processes the
queue when the block has been received.

This behavior is consistent with the [Phase0 specification
](https://github.com/ethereum/consensus-specs/blob/master/specs/phase0/p2p-interface.md#beacon_attestation_subnet_id).

> [IGNORE] The block being voted for
(attestation.data.beacon_block_root) has been seen (via gossip or
non-gossip sources) (a client MAY queue attestations for processing once
block is retrieved).

Once the block being voted for is processed, previously queued
(potentially aggregated) attestations are then processed, and
broadcasted.

Processing (potentially aggregated) attestations takes some non
negligible time. For this reason, (potentially aggregated) attestations
are deduplicated before being introduced into the pending queue, to
avoid eventually processing duplicates.

Before this PR, two aggregated attestations were considered duplicated
if all of the following conditions were gathered:
1. Attestations have the same version, 
2. **Attestations have the same aggregator index (aka., the same
validator aggregated them)**,
3. Attestations have the same slot, 
4. Attestations have the same committee index, and
5. Attestations have the same aggregation bits

Aggregated attestations are then broadcasted.
The final purpose of aggregated attestations is to be packed into the
next block by the next proposer.
When packing attestations, the aggregator index is not used any more.

This pull request modifies the deduplication function used in the
pending aggregated attestations queue by considering that multiple
aggregated attestations only differing by the aggregator index are
equivalent (removing `2.` of the previous list.)

As a consequence, the count of aggregated attestations to be introduced
in the pending queue is reduced from 1 aggregated attestation by
aggregator to, in the best case,
[MAX_COMMITTEE_PER_SLOT=64](https://github.com/ethereum/consensus-specs/blob/master/specs/phase0/beacon-chain.md#misc-1).

Also, only a single aggregated attestation for a given version, slot,
committee index and aggregation bits will be re-broadcasted. This is a
correct behavior, since no data to be included in a block will be lost.
(We can even say that this will reduce by a bit the total networking
volume.)

**How to test**:
1. Start a beacon node (preferably, on a slow computer) from a
checkpoint.
2. Filter logs containing `Synced new block` and `Verified and saved
pending attestations to pool`. (You can pipe logs into `grep -E "Synced
new block|Verified and saved pending attestations to pool"`.

- In `Synced new block` logs, monitor the `sinceSlotStartTime` value.
This should monotonically decrease.
- In `Verified and saved pending attestations to pool`, monitor the
`pendingAggregateAttAndProofCount` value. It should be a "honest" value.
"honest" is not really quantifiable here, since it depends on the
aggregators. But it's likely to be less than
`5*MAX_COMMITTEE_PER_SLOT=320`.

**Which issues(s) does this PR fix?**

Partially fixes:
- https://github.com/OffchainLabs/prysm/issues/16160

**Other notes for review**
Please read commit by commit, with commit messages.
The important commit is b748c04a67.

**Acknowledgements**
- [x] I have read
[CONTRIBUTING.md](https://github.com/prysmaticlabs/prysm/blob/develop/CONTRIBUTING.md).
- [x] I have included a uniquely named [changelog fragment
file](https://github.com/prysmaticlabs/prysm/blob/develop/CONTRIBUTING.md#maintaining-changelogmd).
- [x] I have added a description with sufficient context for reviewers
to understand this PR.
- [x] I have tested that my changes work as expected and I added a
testing plan to the PR description (if applicable).

BUILD.bazel

@@ -193,6 +193,7 @@ nogo(
         "//tools/analyzers/featureconfig:go_default_library",
         "//tools/analyzers/gocognit:go_default_library",
         "//tools/analyzers/ineffassign:go_default_library",
+        "//tools/analyzers/httperror:go_default_library",
         "//tools/analyzers/interfacechecker:go_default_library",
         "//tools/analyzers/logcapitalization:go_default_library",
         "//tools/analyzers/logruswitherror:go_default_library",

beacon-chain/core/helpers/validators.go

@@ -152,7 +152,7 @@ func ActiveValidatorIndices(ctx context.Context, s state.ReadOnlyBeaconState, ep
 	}
 
 	if err := UpdateCommitteeCache(ctx, s, epoch); err != nil {
-		return nil, errors.Wrap(err, "could not update committee cache")
+		log.WithError(err).Error("Could not update committee cache")
 	}
 
 	return indices, nil

beacon-chain/core/peerdas/metrics.go

@@ -5,10 +5,20 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promauto"
 )
 
-var dataColumnComputationTime = promauto.NewHistogram(
-	prometheus.HistogramOpts{
-		Name:    "beacon_data_column_sidecar_computation_milliseconds",
-		Help:    "Captures the time taken to compute data column sidecars from blobs.",
-		Buckets: []float64{25, 50, 100, 250, 500, 750, 1000},
-	},
+var (
+	dataColumnComputationTime = promauto.NewHistogram(
+		prometheus.HistogramOpts{
+			Name:    "beacon_data_column_sidecar_computation_milliseconds",
+			Help:    "Captures the time taken to compute data column sidecars from blobs.",
+			Buckets: []float64{25, 50, 100, 250, 500, 750, 1000},
+		},
+	)
+
+	cellsAndProofsFromStructuredComputationTime = promauto.NewHistogram(
+		prometheus.HistogramOpts{
+			Name:    "cells_and_proofs_from_structured_computation_milliseconds",
+			Help:    "Captures the time taken to compute cells and proofs from structured computation.",
+			Buckets: []float64{10, 20, 30, 40, 50, 100, 200},
+		},
+	)
 )

beacon-chain/core/peerdas/reconstruction.go

@@ -3,6 +3,7 @@ package peerdas
 import (
 	"sort"
 	"sync"
+	"time"
 
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/blockchain/kzg"
 	fieldparams "github.com/OffchainLabs/prysm/v7/config/fieldparams"
@@ -296,32 +297,42 @@ func ComputeCellsAndProofsFromFlat(blobs [][]byte, cellProofs [][]byte) ([][]kzg
 		return nil, nil, ErrBlobsCellsProofsMismatch
 	}
 
-	cellsPerBlob := make([][]kzg.Cell, 0, blobCount)
-	proofsPerBlob := make([][]kzg.Proof, 0, blobCount)
+	var wg errgroup.Group
+
+	cellsPerBlob := make([][]kzg.Cell, blobCount)
+	proofsPerBlob := make([][]kzg.Proof, blobCount)
+
 	for i, blob := range blobs {
-		var kzgBlob kzg.Blob
-		if copy(kzgBlob[:], blob) != len(kzgBlob) {
-			return nil, nil, errors.New("wrong blob size - should never happen")
-		}
-
-		// Compute the extended cells from the (non-extended) blob.
-		cells, err := kzg.ComputeCells(&kzgBlob)
-		if err != nil {
-			return nil, nil, errors.Wrap(err, "compute cells")
-		}
-
-		var proofs []kzg.Proof
-		for idx := uint64(i) * numberOfColumns; idx < (uint64(i)+1)*numberOfColumns; idx++ {
-			var kzgProof kzg.Proof
-			if copy(kzgProof[:], cellProofs[idx]) != len(kzgProof) {
-				return nil, nil, errors.New("wrong KZG proof size - should never happen")
+		wg.Go(func() error {
+			var kzgBlob kzg.Blob
+			if copy(kzgBlob[:], blob) != len(kzgBlob) {
+				return errors.New("wrong blob size - should never happen")
 			}
 
-			proofs = append(proofs, kzgProof)
-		}
+			// Compute the extended cells from the (non-extended) blob.
+			cells, err := kzg.ComputeCells(&kzgBlob)
+			if err != nil {
+				return errors.Wrap(err, "compute cells")
+			}
 
-		cellsPerBlob = append(cellsPerBlob, cells)
-		proofsPerBlob = append(proofsPerBlob, proofs)
+			proofs := make([]kzg.Proof, 0, numberOfColumns)
+			for idx := uint64(i) * numberOfColumns; idx < (uint64(i)+1)*numberOfColumns; idx++ {
+				var kzgProof kzg.Proof
+				if copy(kzgProof[:], cellProofs[idx]) != len(kzgProof) {
+					return errors.New("wrong KZG proof size - should never happen")
+				}
+
+				proofs = append(proofs, kzgProof)
+			}
+
+			cellsPerBlob[i] = cells
+			proofsPerBlob[i] = proofs
+			return nil
+		})
+	}
+
+	if err := wg.Wait(); err != nil {
+		return nil, nil, err
 	}
 
 	return cellsPerBlob, proofsPerBlob, nil
@@ -329,40 +340,55 @@ func ComputeCellsAndProofsFromFlat(blobs [][]byte, cellProofs [][]byte) ([][]kzg
 
 // ComputeCellsAndProofsFromStructured computes the cells and proofs from blobs and cell proofs.
 func ComputeCellsAndProofsFromStructured(blobsAndProofs []*pb.BlobAndProofV2) ([][]kzg.Cell, [][]kzg.Proof, error) {
-	cellsPerBlob := make([][]kzg.Cell, 0, len(blobsAndProofs))
-	proofsPerBlob := make([][]kzg.Proof, 0, len(blobsAndProofs))
-	for _, blobAndProof := range blobsAndProofs {
+	start := time.Now()
+	defer func() {
+		cellsAndProofsFromStructuredComputationTime.Observe(float64(time.Since(start).Milliseconds()))
+	}()
+
+	var wg errgroup.Group
+
+	cellsPerBlob := make([][]kzg.Cell, len(blobsAndProofs))
+	proofsPerBlob := make([][]kzg.Proof, len(blobsAndProofs))
+
+	for i, blobAndProof := range blobsAndProofs {
 		if blobAndProof == nil {
 			return nil, nil, ErrNilBlobAndProof
 		}
 
-		var kzgBlob kzg.Blob
-		if copy(kzgBlob[:], blobAndProof.Blob) != len(kzgBlob) {
-			return nil, nil, errors.New("wrong blob size - should never happen")
-		}
-
-		// Compute the extended cells from the (non-extended) blob.
-		cells, err := kzg.ComputeCells(&kzgBlob)
-		if err != nil {
-			return nil, nil, errors.Wrap(err, "compute cells")
-		}
-
-		kzgProofs := make([]kzg.Proof, 0, fieldparams.NumberOfColumns)
-		for _, kzgProofBytes := range blobAndProof.KzgProofs {
-			if len(kzgProofBytes) != kzg.BytesPerProof {
-				return nil, nil, errors.New("wrong KZG proof size - should never happen")
+		wg.Go(func() error {
+			var kzgBlob kzg.Blob
+			if copy(kzgBlob[:], blobAndProof.Blob) != len(kzgBlob) {
+				return errors.New("wrong blob size - should never happen")
 			}
 
-			var kzgProof kzg.Proof
-			if copy(kzgProof[:], kzgProofBytes) != len(kzgProof) {
-				return nil, nil, errors.New("wrong copied KZG proof size - should never happen")
+			// Compute the extended cells from the (non-extended) blob.
+			cells, err := kzg.ComputeCells(&kzgBlob)
+			if err != nil {
+				return errors.Wrap(err, "compute cells")
 			}
 
-			kzgProofs = append(kzgProofs, kzgProof)
-		}
+			kzgProofs := make([]kzg.Proof, 0, fieldparams.NumberOfColumns)
+			for _, kzgProofBytes := range blobAndProof.KzgProofs {
+				if len(kzgProofBytes) != kzg.BytesPerProof {
+					return errors.New("wrong KZG proof size - should never happen")
+				}
 
-		cellsPerBlob = append(cellsPerBlob, cells)
-		proofsPerBlob = append(proofsPerBlob, kzgProofs)
+				var kzgProof kzg.Proof
+				if copy(kzgProof[:], kzgProofBytes) != len(kzgProof) {
+					return errors.New("wrong copied KZG proof size - should never happen")
+				}
+
+				kzgProofs = append(kzgProofs, kzgProof)
+			}
+
+			cellsPerBlob[i] = cells
+			proofsPerBlob[i] = kzgProofs
+			return nil
+		})
+	}
+
+	if err := wg.Wait(); err != nil {
+		return nil, nil, err
 	}
 
 	return cellsPerBlob, proofsPerBlob, nil

beacon-chain/db/filesystem/data_column.go

@@ -515,6 +515,11 @@ func (dcs *DataColumnStorage) Clear() error {
 
 // prune clean the cache, the filesystem and mutexes.
 func (dcs *DataColumnStorage) prune() {
+	startTime := time.Now()
+	defer func() {
+		dataColumnPruneLatency.Observe(float64(time.Since(startTime).Milliseconds()))
+	}()
+
 	highestStoredEpoch := dcs.cache.HighestEpoch()
 
 	// Check if we need to prune.
@@ -622,6 +627,9 @@ func (dcs *DataColumnStorage) saveDataColumnSidecarsExistingFile(filePath string
 	// Create the SSZ encoded data column sidecars.
 	var sszEncodedDataColumnSidecars []byte
 
+	// Initialize the count of the saved SSZ encoded data column sidecar.
+	storedCount := uint8(0)
+
 	for {
 		dataColumnSidecars := pullChan(inputDataColumnSidecars)
 		if len(dataColumnSidecars) == 0 {
@@ -668,6 +676,9 @@ func (dcs *DataColumnStorage) saveDataColumnSidecarsExistingFile(filePath string
 				return errors.Wrap(err, "set index")
 			}
 
+			// Increment the count of the saved SSZ encoded data column sidecar.
+			storedCount++
+
 			// Append the SSZ encoded data column sidecar to the SSZ encoded data column sidecars.
 			sszEncodedDataColumnSidecars = append(sszEncodedDataColumnSidecars, sszEncodedDataColumnSidecar...)
 		}
@@ -692,9 +703,12 @@ func (dcs *DataColumnStorage) saveDataColumnSidecarsExistingFile(filePath string
 		return errWrongBytesWritten
 	}
 
+	syncStart := time.Now()
 	if err := file.Sync(); err != nil {
 		return errors.Wrap(err, "sync")
 	}
+	dataColumnFileSyncLatency.Observe(float64(time.Since(syncStart).Milliseconds()))
+	dataColumnBatchStoreCount.Observe(float64(storedCount))
 
 	return nil
 }
@@ -808,10 +822,14 @@ func (dcs *DataColumnStorage) saveDataColumnSidecarsNewFile(filePath string, inp
 		return errWrongBytesWritten
 	}
 
+	syncStart := time.Now()
 	if err := file.Sync(); err != nil {
 		return errors.Wrap(err, "sync")
 	}
 
+	dataColumnFileSyncLatency.Observe(float64(time.Since(syncStart).Milliseconds()))
+	dataColumnBatchStoreCount.Observe(float64(storedCount))
+
 	return nil
 }
 

beacon-chain/db/filesystem/metrics.go

@@ -36,16 +36,15 @@ var (
 	})
 
 	// Data columns
-	dataColumnBuckets     = []float64{3, 5, 7, 9, 11, 13}
 	dataColumnSaveLatency = promauto.NewHistogram(prometheus.HistogramOpts{
 		Name:    "data_column_storage_save_latency",
 		Help:    "Latency of DataColumnSidecar storage save operations in milliseconds",
-		Buckets: dataColumnBuckets,
+		Buckets: []float64{10, 20, 30, 50, 100, 200, 500},
 	})
 	dataColumnFetchLatency = promauto.NewHistogram(prometheus.HistogramOpts{
 		Name:    "data_column_storage_get_latency",
 		Help:    "Latency of DataColumnSidecar storage get operations in milliseconds",
-		Buckets: dataColumnBuckets,
+		Buckets: []float64{3, 5, 7, 9, 11, 13},
 	})
 	dataColumnPrunedCounter = promauto.NewCounter(prometheus.CounterOpts{
 		Name: "data_column_pruned",
@@ -59,4 +58,16 @@ var (
 		Name: "data_column_disk_count",
 		Help: "Approximate number of data columns in storage",
 	})
+	dataColumnFileSyncLatency = promauto.NewSummary(prometheus.SummaryOpts{
+		Name: "data_column_file_sync_latency",
+		Help: "Latency of sync operations when saving data columns in milliseconds",
+	})
+	dataColumnBatchStoreCount = promauto.NewSummary(prometheus.SummaryOpts{
+		Name: "data_column_batch_store_count",
+		Help: "Number of data columns stored in a batch",
+	})
+	dataColumnPruneLatency = promauto.NewSummary(prometheus.SummaryOpts{
+		Name: "data_column_prune_latency",
+		Help: "Latency of data column prune operations in milliseconds",
+	})
 )

beacon-chain/execution/engine_client.go

@@ -532,12 +532,19 @@ func (s *Service) GetBlobsV2(ctx context.Context, versionedHashes []common.Hash)
 	ctx, span := trace.StartSpan(ctx, "powchain.engine-api-client.GetBlobsV2")
 	defer span.End()
 
+	start := time.Now()
+
 	if !s.capabilityCache.has(GetBlobsV2) {
 		return nil, errors.New(fmt.Sprintf("%s is not supported", GetBlobsV2))
 	}
 
 	result := make([]*pb.BlobAndProofV2, len(versionedHashes))
 	err := s.rpcClient.CallContext(ctx, &result, GetBlobsV2, versionedHashes)
+
+	if len(result) != 0 {
+		getBlobsV2Latency.Observe(float64(time.Since(start).Milliseconds()))
+	}
+
 	return result, handleRPCError(err)
 }
 

beacon-chain/execution/metrics.go

@@ -27,6 +27,13 @@ var (
 			Buckets: []float64{25, 50, 100, 200, 500, 1000, 2000, 4000},
 		},
 	)
+	getBlobsV2Latency = promauto.NewHistogram(
+		prometheus.HistogramOpts{
+			Name:    "get_blobs_v2_latency_milliseconds",
+			Help:    "Captures RPC latency for getBlobsV2 in milliseconds",
+			Buckets: []float64{25, 50, 100, 200, 500, 1000, 2000, 4000},
+		},
+	)
 	errParseCount = promauto.NewCounter(prometheus.CounterOpts{
 		Name: "execution_parse_error_count",
 		Help: "The number of errors that occurred while parsing execution payload",

beacon-chain/forkchoice/doubly-linked-tree/unrealized_justification.go

@@ -88,9 +88,6 @@ func (s *Store) pullTips(state state.BeaconState, node *Node, jc, fc *ethpb.Chec
 		}
 	}
 	if uf.Epoch > s.unrealizedFinalizedCheckpoint.Epoch {
-		s.unrealizedJustifiedCheckpoint = &forkchoicetypes.Checkpoint{
-			Epoch: uj.Epoch, Root: bytesutil.ToBytes32(uj.Root),
-		}
 		s.unrealizedFinalizedCheckpoint = &forkchoicetypes.Checkpoint{
 			Epoch: uf.Epoch, Root: bytesutil.ToBytes32(uf.Root),
 		}

beacon-chain/p2p/types/object_mapping.go

@@ -204,6 +204,9 @@ func InitializeDataMaps() {
 		bytesutil.ToBytes4(params.BeaconConfig().ElectraForkVersion): func() (interfaces.LightClientOptimisticUpdate, error) {
 			return lightclientConsensusTypes.NewEmptyOptimisticUpdateDeneb(), nil
 		},
+		bytesutil.ToBytes4(params.BeaconConfig().FuluForkVersion): func() (interfaces.LightClientOptimisticUpdate, error) {
+			return lightclientConsensusTypes.NewEmptyOptimisticUpdateDeneb(), nil
+		},
 	}
 
 	// Reset our light client finality update map.
@@ -223,5 +226,8 @@ func InitializeDataMaps() {
 		bytesutil.ToBytes4(params.BeaconConfig().ElectraForkVersion): func() (interfaces.LightClientFinalityUpdate, error) {
 			return lightclientConsensusTypes.NewEmptyFinalityUpdateElectra(), nil
 		},
+		bytesutil.ToBytes4(params.BeaconConfig().FuluForkVersion): func() (interfaces.LightClientFinalityUpdate, error) {
+			return lightclientConsensusTypes.NewEmptyFinalityUpdateElectra(), nil
+		},
 	}
 }

beacon-chain/rpc/eth/beacon/handlers_pool.go

@@ -130,6 +130,10 @@ func (s *Server) SubmitAttestationsV2(w http.ResponseWriter, r *http.Request) {
 	ctx, span := trace.StartSpan(r.Context(), "beacon.SubmitAttestationsV2")
 	defer span.End()
 
+	if shared.IsSyncing(ctx, w, s.SyncChecker, s.HeadFetcher, s.TimeFetcher, s.OptimisticModeFetcher) {
+		return
+	}
+
 	versionHeader := r.Header.Get(api.VersionHeader)
 	if versionHeader == "" {
 		httputil.HandleError(w, api.VersionHeader+" header is required", http.StatusBadRequest)
@@ -238,22 +242,14 @@ func (s *Server) handleAttestationsElectra(
 			},
 		})
 
-		targetState, err := s.AttestationStateFetcher.AttestationTargetState(ctx, singleAtt.Data.Target)
-		if err != nil {
-			return nil, nil, errors.Wrap(err, "could not get target state for attestation")
-		}
-		committee, err := corehelpers.BeaconCommitteeFromState(ctx, targetState, singleAtt.Data.Slot, singleAtt.CommitteeId)
-		if err != nil {
-			return nil, nil, errors.Wrap(err, "could not get committee for attestation")
-		}
-		att := singleAtt.ToAttestationElectra(committee)
-
-		wantedEpoch := slots.ToEpoch(att.Data.Slot)
+		// Broadcast first using CommitteeId directly (fast path)
+		// This matches gRPC behavior and avoids blocking on state fetching
+		wantedEpoch := slots.ToEpoch(singleAtt.Data.Slot)
 		vals, err := s.HeadFetcher.HeadValidatorsIndices(ctx, wantedEpoch)
 		if err != nil {
 			return nil, nil, errors.Wrap(err, "could not get head validator indices")
 		}
-		subnet := corehelpers.ComputeSubnetFromCommitteeAndSlot(uint64(len(vals)), att.GetCommitteeIndex(), att.Data.Slot)
+		subnet := corehelpers.ComputeSubnetFromCommitteeAndSlot(uint64(len(vals)), singleAtt.CommitteeId, singleAtt.Data.Slot)
 		if err = s.Broadcaster.BroadcastAttestation(ctx, subnet, singleAtt); err != nil {
 			failedBroadcasts = append(failedBroadcasts, &server.IndexedError{
 				Index:   i,
@@ -264,17 +260,35 @@ func (s *Server) handleAttestationsElectra(
 			}
 			continue
 		}
+	}
 
-		if features.Get().EnableExperimentalAttestationPool {
-			if err = s.AttestationCache.Add(att); err != nil {
-				log.WithError(err).Error("Could not save attestation")
+	// Save to pool after broadcast (slow path - requires state fetching)
+	// Run in goroutine to avoid blocking the HTTP response
+	go func() {
+		for _, singleAtt := range validAttestations {
+			targetState, err := s.AttestationStateFetcher.AttestationTargetState(context.Background(), singleAtt.Data.Target)
+			if err != nil {
+				log.WithError(err).Error("Could not get target state for attestation")
+				continue
 			}
-		} else {
-			if err = s.AttestationsPool.SaveUnaggregatedAttestation(att); err != nil {
-				log.WithError(err).Error("Could not save attestation")
+			committee, err := corehelpers.BeaconCommitteeFromState(context.Background(), targetState, singleAtt.Data.Slot, singleAtt.CommitteeId)
+			if err != nil {
+				log.WithError(err).Error("Could not get committee for attestation")
+				continue
+			}
+			att := singleAtt.ToAttestationElectra(committee)
+
+			if features.Get().EnableExperimentalAttestationPool {
+				if err = s.AttestationCache.Add(att); err != nil {
+					log.WithError(err).Error("Could not save attestation")
+				}
+			} else {
+				if err = s.AttestationsPool.SaveUnaggregatedAttestation(att); err != nil {
+					log.WithError(err).Error("Could not save attestation")
+				}
 			}
 		}
-	}
+	}()
 
 	if len(failedBroadcasts) > 0 {
 		log.WithFields(logrus.Fields{
@@ -470,6 +484,10 @@ func (s *Server) SubmitSyncCommitteeSignatures(w http.ResponseWriter, r *http.Re
 	ctx, span := trace.StartSpan(r.Context(), "beacon.SubmitPoolSyncCommitteeSignatures")
 	defer span.End()
 
+	if shared.IsSyncing(ctx, w, s.SyncChecker, s.HeadFetcher, s.TimeFetcher, s.OptimisticModeFetcher) {
+		return
+	}
+
 	var req structs.SubmitSyncCommitteeSignaturesRequest
 	err := json.NewDecoder(r.Body).Decode(&req.Data)
 	switch {
@@ -711,6 +729,7 @@ func (s *Server) SubmitAttesterSlashingsV2(w http.ResponseWriter, r *http.Reques
 	versionHeader := r.Header.Get(api.VersionHeader)
 	if versionHeader == "" {
 		httputil.HandleError(w, api.VersionHeader+" header is required", http.StatusBadRequest)
+		return
 	}
 	v, err := version.FromString(versionHeader)
 	if err != nil {

beacon-chain/rpc/eth/beacon/handlers_pool_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/operations/voluntaryexits/mock"
 	p2pMock "github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/testing"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/rpc/core"
+	mockSync "github.com/OffchainLabs/prysm/v7/beacon-chain/sync/initial-sync/testing"
 	state_native "github.com/OffchainLabs/prysm/v7/beacon-chain/state/state-native"
 	"github.com/OffchainLabs/prysm/v7/config/params"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
@@ -622,6 +623,8 @@ func TestSubmitAttestationsV2(t *testing.T) {
 		HeadFetcher:             chainService,
 		ChainInfoFetcher:        chainService,
 		TimeFetcher:             chainService,
+		OptimisticModeFetcher:   chainService,
+		SyncChecker:             &mockSync.Sync{IsSyncing: false},
 		OperationNotifier:       &blockchainmock.MockOperationNotifier{},
 		AttestationStateFetcher: chainService,
 	}
@@ -654,6 +657,7 @@ func TestSubmitAttestationsV2(t *testing.T) {
 			assert.Equal(t, primitives.Epoch(0), broadcaster.BroadcastAttestations[0].GetData().Source.Epoch)
 			assert.Equal(t, "0xcf8e0d4e9587369b2301d0790347320302cc0943d5a1884560367e8208d920f2", hexutil.Encode(broadcaster.BroadcastAttestations[0].GetData().Target.Root))
 			assert.Equal(t, primitives.Epoch(0), broadcaster.BroadcastAttestations[0].GetData().Target.Epoch)
+			time.Sleep(100 * time.Millisecond) // Wait for async pool save
 			assert.Equal(t, 1, s.AttestationsPool.UnaggregatedAttestationCount())
 		})
 		t.Run("multiple", func(t *testing.T) {
@@ -673,6 +677,7 @@ func TestSubmitAttestationsV2(t *testing.T) {
 			assert.Equal(t, http.StatusOK, writer.Code)
 			assert.Equal(t, true, broadcaster.BroadcastCalled.Load())
 			assert.Equal(t, 2, broadcaster.NumAttestations())
+			time.Sleep(100 * time.Millisecond) // Wait for async pool save
 			assert.Equal(t, 2, s.AttestationsPool.UnaggregatedAttestationCount())
 		})
 		t.Run("phase0 att post electra", func(t *testing.T) {
@@ -793,6 +798,7 @@ func TestSubmitAttestationsV2(t *testing.T) {
 			assert.Equal(t, primitives.Epoch(0), broadcaster.BroadcastAttestations[0].GetData().Source.Epoch)
 			assert.Equal(t, "0xcf8e0d4e9587369b2301d0790347320302cc0943d5a1884560367e8208d920f2", hexutil.Encode(broadcaster.BroadcastAttestations[0].GetData().Target.Root))
 			assert.Equal(t, primitives.Epoch(0), broadcaster.BroadcastAttestations[0].GetData().Target.Epoch)
+			time.Sleep(100 * time.Millisecond) // Wait for async pool save
 			assert.Equal(t, 1, s.AttestationsPool.UnaggregatedAttestationCount())
 		})
 		t.Run("multiple", func(t *testing.T) {
@@ -812,6 +818,7 @@ func TestSubmitAttestationsV2(t *testing.T) {
 			assert.Equal(t, http.StatusOK, writer.Code)
 			assert.Equal(t, true, broadcaster.BroadcastCalled.Load())
 			assert.Equal(t, 2, broadcaster.NumAttestations())
+			time.Sleep(100 * time.Millisecond) // Wait for async pool save
 			assert.Equal(t, 2, s.AttestationsPool.UnaggregatedAttestationCount())
 		})
 		t.Run("no body", func(t *testing.T) {
@@ -861,6 +868,27 @@ func TestSubmitAttestationsV2(t *testing.T) {
 			assert.Equal(t, true, strings.Contains(e.Failures[0].Message, "Incorrect attestation signature"))
 		})
 	})
+	t.Run("syncing", func(t *testing.T) {
+		chainService := &blockchainmock.ChainService{}
+		s := &Server{
+			HeadFetcher:           chainService,
+			TimeFetcher:           chainService,
+			OptimisticModeFetcher: chainService,
+			SyncChecker:           &mockSync.Sync{IsSyncing: true},
+		}
+
+		var body bytes.Buffer
+		_, err := body.WriteString(singleAtt)
+		require.NoError(t, err)
+		request := httptest.NewRequest(http.MethodPost, "http://example.com", &body)
+		request.Header.Set(api.VersionHeader, version.String(version.Phase0))
+		writer := httptest.NewRecorder()
+		writer.Body = &bytes.Buffer{}
+
+		s.SubmitAttestationsV2(writer, request)
+		assert.Equal(t, http.StatusServiceUnavailable, writer.Code)
+		assert.Equal(t, true, strings.Contains(writer.Body.String(), "Beacon node is currently syncing"))
+	})
 }
 
 func TestListVoluntaryExits(t *testing.T) {
@@ -1057,14 +1085,19 @@ func TestSubmitSyncCommitteeSignatures(t *testing.T) {
 
 	t.Run("single", func(t *testing.T) {
 		broadcaster := &p2pMock.MockBroadcaster{}
+		chainService := &blockchainmock.ChainService{
+			State:                st,
+			SyncCommitteeIndices: []primitives.CommitteeIndex{0},
+		}
 		s := &Server{
+			HeadFetcher:           chainService,
+			TimeFetcher:           chainService,
+			OptimisticModeFetcher: chainService,
+			SyncChecker:           &mockSync.Sync{IsSyncing: false},
 			CoreService: &core.Service{
 				SyncCommitteePool: synccommittee.NewStore(),
 				P2P:               broadcaster,
-				HeadFetcher: &blockchainmock.ChainService{
-					State:                st,
-					SyncCommitteeIndices: []primitives.CommitteeIndex{0},
-				},
+				HeadFetcher:       chainService,
 			},
 		}
 
@@ -1089,14 +1122,19 @@ func TestSubmitSyncCommitteeSignatures(t *testing.T) {
 	})
 	t.Run("multiple", func(t *testing.T) {
 		broadcaster := &p2pMock.MockBroadcaster{}
+		chainService := &blockchainmock.ChainService{
+			State:                st,
+			SyncCommitteeIndices: []primitives.CommitteeIndex{0},
+		}
 		s := &Server{
+			HeadFetcher:           chainService,
+			TimeFetcher:           chainService,
+			OptimisticModeFetcher: chainService,
+			SyncChecker:           &mockSync.Sync{IsSyncing: false},
 			CoreService: &core.Service{
 				SyncCommitteePool: synccommittee.NewStore(),
 				P2P:               broadcaster,
-				HeadFetcher: &blockchainmock.ChainService{
-					State:                st,
-					SyncCommitteeIndices: []primitives.CommitteeIndex{0},
-				},
+				HeadFetcher:       chainService,
 			},
 		}
 
@@ -1120,13 +1158,18 @@ func TestSubmitSyncCommitteeSignatures(t *testing.T) {
 	})
 	t.Run("invalid", func(t *testing.T) {
 		broadcaster := &p2pMock.MockBroadcaster{}
+		chainService := &blockchainmock.ChainService{
+			State: st,
+		}
 		s := &Server{
+			HeadFetcher:           chainService,
+			TimeFetcher:           chainService,
+			OptimisticModeFetcher: chainService,
+			SyncChecker:           &mockSync.Sync{IsSyncing: false},
 			CoreService: &core.Service{
 				SyncCommitteePool: synccommittee.NewStore(),
 				P2P:               broadcaster,
-				HeadFetcher: &blockchainmock.ChainService{
-					State: st,
-				},
+				HeadFetcher:       chainService,
 			},
 		}
 
@@ -1149,7 +1192,13 @@ func TestSubmitSyncCommitteeSignatures(t *testing.T) {
 		assert.Equal(t, false, broadcaster.BroadcastCalled.Load())
 	})
 	t.Run("empty", func(t *testing.T) {
-		s := &Server{}
+		chainService := &blockchainmock.ChainService{State: st}
+		s := &Server{
+			HeadFetcher:           chainService,
+			TimeFetcher:           chainService,
+			OptimisticModeFetcher: chainService,
+			SyncChecker:           &mockSync.Sync{IsSyncing: false},
+		}
 
 		var body bytes.Buffer
 		_, err := body.WriteString("[]")
@@ -1166,7 +1215,13 @@ func TestSubmitSyncCommitteeSignatures(t *testing.T) {
 		assert.Equal(t, true, strings.Contains(e.Message, "No data submitted"))
 	})
 	t.Run("no body", func(t *testing.T) {
-		s := &Server{}
+		chainService := &blockchainmock.ChainService{State: st}
+		s := &Server{
+			HeadFetcher:           chainService,
+			TimeFetcher:           chainService,
+			OptimisticModeFetcher: chainService,
+			SyncChecker:           &mockSync.Sync{IsSyncing: false},
+		}
 
 		request := httptest.NewRequest(http.MethodPost, "http://example.com", nil)
 		writer := httptest.NewRecorder()
@@ -1179,6 +1234,26 @@ func TestSubmitSyncCommitteeSignatures(t *testing.T) {
 		assert.Equal(t, http.StatusBadRequest, e.Code)
 		assert.Equal(t, true, strings.Contains(e.Message, "No data submitted"))
 	})
+	t.Run("syncing", func(t *testing.T) {
+		chainService := &blockchainmock.ChainService{State: st}
+		s := &Server{
+			HeadFetcher:           chainService,
+			TimeFetcher:           chainService,
+			OptimisticModeFetcher: chainService,
+			SyncChecker:           &mockSync.Sync{IsSyncing: true},
+		}
+
+		var body bytes.Buffer
+		_, err := body.WriteString(singleSyncCommitteeMsg)
+		require.NoError(t, err)
+		request := httptest.NewRequest(http.MethodPost, "http://example.com", &body)
+		writer := httptest.NewRecorder()
+		writer.Body = &bytes.Buffer{}
+
+		s.SubmitSyncCommitteeSignatures(writer, request)
+		assert.Equal(t, http.StatusServiceUnavailable, writer.Code)
+		assert.Equal(t, true, strings.Contains(writer.Body.String(), "Beacon node is currently syncing"))
+	})
 }
 
 func TestListBLSToExecutionChanges(t *testing.T) {
@@ -2112,6 +2187,33 @@ func TestSubmitAttesterSlashingsV2(t *testing.T) {
 		assert.Equal(t, http.StatusBadRequest, e.Code)
 		assert.StringContains(t, "Invalid attester slashing", e.Message)
 	})
+
+	t.Run("missing-version-header", func(t *testing.T) {
+		bs, err := util.NewBeaconStateElectra()
+		require.NoError(t, err)
+
+		broadcaster := &p2pMock.MockBroadcaster{}
+		s := &Server{
+			ChainInfoFetcher: &blockchainmock.ChainService{State: bs},
+			SlashingsPool:    &slashingsmock.PoolMock{},
+			Broadcaster:      broadcaster,
+		}
+
+		var body bytes.Buffer
+		_, err = body.WriteString(invalidAttesterSlashing)
+		require.NoError(t, err)
+		request := httptest.NewRequest(http.MethodPost, "http://example.com/beacon/pool/attester_slashings", &body)
+		// Intentionally do not set api.VersionHeader to verify missing header handling.
+		writer := httptest.NewRecorder()
+		writer.Body = &bytes.Buffer{}
+
+		s.SubmitAttesterSlashingsV2(writer, request)
+		require.Equal(t, http.StatusBadRequest, writer.Code)
+		e := &httputil.DefaultJsonError{}
+		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
+		assert.Equal(t, http.StatusBadRequest, e.Code)
+		assert.StringContains(t, api.VersionHeader+" header is required", e.Message)
+	})
 }
 
 func TestSubmitProposerSlashing_InvalidSlashing(t *testing.T) {

beacon-chain/rpc/eth/beacon/handlers_state_test.go

@@ -654,6 +654,10 @@ func (m *futureSyncMockFetcher) StateBySlot(context.Context, primitives.Slot) (s
 	return m.BeaconState, nil
 }
 
+func (m *futureSyncMockFetcher) StateByEpoch(context.Context, primitives.Epoch) (state.BeaconState, error) {
+	return m.BeaconState, nil
+}
+
 func TestGetSyncCommittees_Future(t *testing.T) {
 	st, _ := util.DeterministicGenesisStateAltair(t, params.BeaconConfig().SyncCommitteeSize)
 	syncCommittee := make([][]byte, params.BeaconConfig().SyncCommitteeSize)

beacon-chain/rpc/eth/light-client/handlers.go

@@ -116,6 +116,7 @@ func (s *Server) GetLightClientUpdatesByRange(w http.ResponseWriter, req *http.R
 		for _, update := range updates {
 			if ctx.Err() != nil {
 				httputil.HandleError(w, "Context error: "+ctx.Err().Error(), http.StatusInternalServerError)
+				return
 			}
 
 			updateSlot := update.AttestedHeader().Beacon().Slot
@@ -131,12 +132,15 @@ func (s *Server) GetLightClientUpdatesByRange(w http.ResponseWriter, req *http.R
 			chunkLength = ssz.MarshalUint64(chunkLength, uint64(len(updateSSZ)+4))
 			if _, err := w.Write(chunkLength); err != nil {
 				httputil.HandleError(w, "Could not write chunk length: "+err.Error(), http.StatusInternalServerError)
+				return
 			}
 			if _, err := w.Write(updateEntry.ForkDigest[:]); err != nil {
 				httputil.HandleError(w, "Could not write fork digest: "+err.Error(), http.StatusInternalServerError)
+				return
 			}
 			if _, err := w.Write(updateSSZ); err != nil {
 				httputil.HandleError(w, "Could not write update SSZ: "+err.Error(), http.StatusInternalServerError)
+				return
 			}
 		}
 	} else {
@@ -145,6 +149,7 @@ func (s *Server) GetLightClientUpdatesByRange(w http.ResponseWriter, req *http.R
 		for _, update := range updates {
 			if ctx.Err() != nil {
 				httputil.HandleError(w, "Context error: "+ctx.Err().Error(), http.StatusInternalServerError)
+				return
 			}
 
 			updateJson, err := structs.LightClientUpdateFromConsensus(update)

beacon-chain/rpc/eth/node/handlers.go

@@ -132,6 +132,7 @@ func (s *Server) GetHealth(w http.ResponseWriter, r *http.Request) {
 	optimistic, err := s.OptimisticModeFetcher.IsOptimistic(ctx)
 	if err != nil {
 		httputil.HandleError(w, "Could not check optimistic status: "+err.Error(), http.StatusInternalServerError)
+		return
 	}
 	if s.SyncChecker.Synced() && !optimistic {
 		return

beacon-chain/rpc/eth/rewards/handlers.go

@@ -228,7 +228,7 @@ func (s *Server) attRewardsState(w http.ResponseWriter, r *http.Request) (state.
 	}
 	st, err := s.Stater.StateBySlot(r.Context(), nextEpochEnd)
 	if err != nil {
-		httputil.HandleError(w, "Could not get state for epoch's starting slot: "+err.Error(), http.StatusInternalServerError)
+		shared.WriteStateFetchError(w, err)
 		return nil, false
 	}
 	return st, true

beacon-chain/rpc/eth/validator/BUILD.bazel

@@ -19,7 +19,6 @@ go_library(
         "//beacon-chain/cache:go_default_library",
         "//beacon-chain/core/feed/operation:go_default_library",
         "//beacon-chain/core/helpers:go_default_library",
-        "//beacon-chain/core/transition:go_default_library",
         "//beacon-chain/db:go_default_library",
         "//beacon-chain/operations/attestations:go_default_library",
         "//beacon-chain/operations/synccommittee:go_default_library",
@@ -78,6 +77,7 @@ go_test(
         "//beacon-chain/rpc/core:go_default_library",
         "//beacon-chain/rpc/eth/rewards/testing:go_default_library",
         "//beacon-chain/rpc/eth/shared/testing:go_default_library",
+        "//beacon-chain/rpc/lookup:go_default_library",
         "//beacon-chain/rpc/testutil:go_default_library",
         "//beacon-chain/state:go_default_library",
         "//beacon-chain/state/stategen:go_default_library",

beacon-chain/rpc/eth/validator/handlers.go

@@ -19,7 +19,6 @@ import (
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/builder"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/cache"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/helpers"
-	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/transition"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/rpc/core"
 	rpchelpers "github.com/OffchainLabs/prysm/v7/beacon-chain/rpc/eth/helpers"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/rpc/eth/shared"
@@ -898,20 +897,15 @@ func (s *Server) GetAttesterDuties(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	var startSlot primitives.Slot
+	// For next epoch requests, we use the current epoch's state since committee
+	// assignments for next epoch can be computed from current epoch's state.
+	epochForState := requestedEpoch
 	if requestedEpoch == nextEpoch {
-		startSlot, err = slots.EpochStart(currentEpoch)
-	} else {
-		startSlot, err = slots.EpochStart(requestedEpoch)
+		epochForState = currentEpoch
 	}
+	st, err := s.Stater.StateByEpoch(ctx, epochForState)
 	if err != nil {
-		httputil.HandleError(w, fmt.Sprintf("Could not get start slot from epoch %d: %v", requestedEpoch, err), http.StatusInternalServerError)
-		return
-	}
-
-	st, err := s.Stater.StateBySlot(ctx, startSlot)
-	if err != nil {
-		httputil.HandleError(w, "Could not get state: "+err.Error(), http.StatusInternalServerError)
+		shared.WriteStateFetchError(w, err)
 		return
 	}
 
@@ -1020,39 +1014,11 @@ func (s *Server) GetProposerDuties(w http.ResponseWriter, r *http.Request) {
 		nextEpochLookahead = true
 	}
 
-	epochStartSlot, err := slots.EpochStart(requestedEpoch)
+	st, err := s.Stater.StateByEpoch(ctx, requestedEpoch)
 	if err != nil {
-		httputil.HandleError(w, fmt.Sprintf("Could not get start slot of epoch %d: %v", requestedEpoch, err), http.StatusInternalServerError)
+		shared.WriteStateFetchError(w, err)
 		return
 	}
-	var st state.BeaconState
-	// if the requested epoch is new, use the head state and the next slot cache
-	if requestedEpoch < currentEpoch {
-		st, err = s.Stater.StateBySlot(ctx, epochStartSlot)
-		if err != nil {
-			httputil.HandleError(w, fmt.Sprintf("Could not get state for slot %d: %v ", epochStartSlot, err), http.StatusInternalServerError)
-			return
-		}
-	} else {
-		st, err = s.HeadFetcher.HeadState(ctx)
-		if err != nil {
-			httputil.HandleError(w, fmt.Sprintf("Could not get head state: %v ", err), http.StatusInternalServerError)
-			return
-		}
-		// Notice that even for Fulu requests for the next epoch, we are only advancing the state to the start of the current epoch.
-		if st.Slot() < epochStartSlot {
-			headRoot, err := s.HeadFetcher.HeadRoot(ctx)
-			if err != nil {
-				httputil.HandleError(w, fmt.Sprintf("Could not get head root: %v ", err), http.StatusInternalServerError)
-				return
-			}
-			st, err = transition.ProcessSlotsUsingNextSlotCache(ctx, st, headRoot, epochStartSlot)
-			if err != nil {
-				httputil.HandleError(w, fmt.Sprintf("Could not process slots up to %d: %v ", epochStartSlot, err), http.StatusInternalServerError)
-				return
-			}
-		}
-	}
 
 	var assignments map[primitives.ValidatorIndex][]primitives.Slot
 	if nextEpochLookahead {
@@ -1103,7 +1069,8 @@ func (s *Server) GetProposerDuties(w http.ResponseWriter, r *http.Request) {
 		httputil.HandleError(w, "Could not check optimistic status: "+err.Error(), http.StatusInternalServerError)
 		return
 	}
-	if !sortProposerDuties(w, duties) {
+	if err = sortProposerDuties(duties); err != nil {
+		httputil.HandleError(w, "Could not sort proposer duties: "+err.Error(), http.StatusInternalServerError)
 		return
 	}
 
@@ -1174,14 +1141,10 @@ func (s *Server) GetSyncCommitteeDuties(w http.ResponseWriter, r *http.Request)
 	}
 
 	startingEpoch := min(requestedEpoch, currentEpoch)
-	slot, err := slots.EpochStart(startingEpoch)
+
+	st, err := s.Stater.StateByEpoch(ctx, startingEpoch)
 	if err != nil {
-		httputil.HandleError(w, "Could not get sync committee slot: "+err.Error(), http.StatusInternalServerError)
-		return
-	}
-	st, err := s.Stater.State(ctx, []byte(strconv.FormatUint(uint64(slot), 10)))
-	if err != nil {
-		httputil.HandleError(w, "Could not get sync committee state: "+err.Error(), http.StatusInternalServerError)
+		shared.WriteStateFetchError(w, err)
 		return
 	}
 
@@ -1327,7 +1290,7 @@ func (s *Server) GetLiveness(w http.ResponseWriter, r *http.Request) {
 		}
 		st, err = s.Stater.StateBySlot(ctx, epochEnd)
 		if err != nil {
-			httputil.HandleError(w, "Could not get slot for requested epoch: "+err.Error(), http.StatusInternalServerError)
+			shared.WriteStateFetchError(w, err)
 			return
 		}
 		participation, err = st.CurrentEpochParticipation()
@@ -1447,22 +1410,20 @@ func syncCommitteeDutiesAndVals(
 	return duties, vals, nil
 }
 
-func sortProposerDuties(w http.ResponseWriter, duties []*structs.ProposerDuty) bool {
-	ok := true
+func sortProposerDuties(duties []*structs.ProposerDuty) error {
+	var err error
 	sort.Slice(duties, func(i, j int) bool {
-		si, err := strconv.ParseUint(duties[i].Slot, 10, 64)
-		if err != nil {
-			httputil.HandleError(w, "Could not parse slot: "+err.Error(), http.StatusInternalServerError)
-			ok = false
+		si, parseErr := strconv.ParseUint(duties[i].Slot, 10, 64)
+		if parseErr != nil {
+			err = errors.Wrap(parseErr, "could not parse slot")
 			return false
 		}
-		sj, err := strconv.ParseUint(duties[j].Slot, 10, 64)
-		if err != nil {
-			httputil.HandleError(w, "Could not parse slot: "+err.Error(), http.StatusInternalServerError)
-			ok = false
+		sj, parseErr := strconv.ParseUint(duties[j].Slot, 10, 64)
+		if parseErr != nil {
+			err = errors.Wrap(parseErr, "could not parse slot")
 			return false
 		}
 		return si < sj
 	})
-	return ok
+	return err
 }

beacon-chain/rpc/eth/validator/handlers_test.go

@@ -25,6 +25,7 @@ import (
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/operations/synccommittee"
 	p2pmock "github.com/OffchainLabs/prysm/v7/beacon-chain/p2p/testing"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/rpc/core"
+	"github.com/OffchainLabs/prysm/v7/beacon-chain/rpc/lookup"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/rpc/testutil"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/state"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/state/stategen"
@@ -2006,6 +2007,7 @@ func TestGetAttesterDuties(t *testing.T) {
 		TimeFetcher:           chain,
 		SyncChecker:           &mockSync.Sync{IsSyncing: false},
 		OptimisticModeFetcher: chain,
+		HeadFetcher:           chain,
 		BeaconDB:              db,
 	}
 
@@ -2184,6 +2186,7 @@ func TestGetAttesterDuties(t *testing.T) {
 			Stater:                &testutil.MockStater{StatesBySlot: map[primitives.Slot]state.BeaconState{0: bs}},
 			TimeFetcher:           chain,
 			OptimisticModeFetcher: chain,
+			HeadFetcher:           chain,
 			SyncChecker:           &mockSync.Sync{IsSyncing: false},
 			BeaconDB:              db,
 		}
@@ -2224,6 +2227,62 @@ func TestGetAttesterDuties(t *testing.T) {
 		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
 		assert.Equal(t, http.StatusServiceUnavailable, e.Code)
 	})
+	t.Run("state not found returns 404", func(t *testing.T) {
+		chainSlot := primitives.Slot(0)
+		chain := &mockChain.ChainService{
+			State: bs, Root: genesisRoot[:], Slot: &chainSlot,
+		}
+		stateNotFoundErr := lookup.NewStateNotFoundError(8192, []byte("test"))
+		s := &Server{
+			Stater:                &testutil.MockStater{CustomError: &stateNotFoundErr},
+			TimeFetcher:           chain,
+			SyncChecker:           &mockSync.Sync{IsSyncing: false},
+			OptimisticModeFetcher: chain,
+			HeadFetcher:           chain,
+		}
+
+		var body bytes.Buffer
+		_, err = body.WriteString("[\"0\"]")
+		require.NoError(t, err)
+		request := httptest.NewRequest(http.MethodGet, "http://www.example.com/eth/v1/validator/duties/attester/{epoch}", &body)
+		request.SetPathValue("epoch", "0")
+		writer := httptest.NewRecorder()
+		writer.Body = &bytes.Buffer{}
+
+		s.GetAttesterDuties(writer, request)
+		assert.Equal(t, http.StatusNotFound, writer.Code)
+		e := &httputil.DefaultJsonError{}
+		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
+		assert.Equal(t, http.StatusNotFound, e.Code)
+		assert.StringContains(t, "State not found", e.Message)
+	})
+	t.Run("state fetch error returns 500", func(t *testing.T) {
+		chainSlot := primitives.Slot(0)
+		chain := &mockChain.ChainService{
+			State: bs, Root: genesisRoot[:], Slot: &chainSlot,
+		}
+		s := &Server{
+			Stater:                &testutil.MockStater{CustomError: errors.New("internal error")},
+			TimeFetcher:           chain,
+			SyncChecker:           &mockSync.Sync{IsSyncing: false},
+			OptimisticModeFetcher: chain,
+			HeadFetcher:           chain,
+		}
+
+		var body bytes.Buffer
+		_, err = body.WriteString("[\"0\"]")
+		require.NoError(t, err)
+		request := httptest.NewRequest(http.MethodGet, "http://www.example.com/eth/v1/validator/duties/attester/{epoch}", &body)
+		request.SetPathValue("epoch", "0")
+		writer := httptest.NewRecorder()
+		writer.Body = &bytes.Buffer{}
+
+		s.GetAttesterDuties(writer, request)
+		assert.Equal(t, http.StatusInternalServerError, writer.Code)
+		e := &httputil.DefaultJsonError{}
+		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
+		assert.Equal(t, http.StatusInternalServerError, e.Code)
+	})
 }
 
 func TestGetProposerDuties(t *testing.T) {
@@ -2427,6 +2486,60 @@ func TestGetProposerDuties(t *testing.T) {
 		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
 		assert.Equal(t, http.StatusServiceUnavailable, e.Code)
 	})
+	t.Run("state not found returns 404", func(t *testing.T) {
+		bs, err := transition.GenesisBeaconState(t.Context(), deposits, 0, eth1Data)
+		require.NoError(t, err)
+		chainSlot := primitives.Slot(0)
+		chain := &mockChain.ChainService{
+			State: bs, Root: genesisRoot[:], Slot: &chainSlot,
+		}
+		stateNotFoundErr := lookup.NewStateNotFoundError(8192, []byte("test"))
+		s := &Server{
+			Stater:                &testutil.MockStater{CustomError: &stateNotFoundErr},
+			TimeFetcher:           chain,
+			SyncChecker:           &mockSync.Sync{IsSyncing: false},
+			OptimisticModeFetcher: chain,
+			HeadFetcher:           chain,
+		}
+
+		request := httptest.NewRequest(http.MethodGet, "http://www.example.com/eth/v1/validator/duties/proposer/{epoch}", nil)
+		request.SetPathValue("epoch", "0")
+		writer := httptest.NewRecorder()
+		writer.Body = &bytes.Buffer{}
+
+		s.GetProposerDuties(writer, request)
+		assert.Equal(t, http.StatusNotFound, writer.Code)
+		e := &httputil.DefaultJsonError{}
+		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
+		assert.Equal(t, http.StatusNotFound, e.Code)
+		assert.StringContains(t, "State not found", e.Message)
+	})
+	t.Run("state fetch error returns 500", func(t *testing.T) {
+		bs, err := transition.GenesisBeaconState(t.Context(), deposits, 0, eth1Data)
+		require.NoError(t, err)
+		chainSlot := primitives.Slot(0)
+		chain := &mockChain.ChainService{
+			State: bs, Root: genesisRoot[:], Slot: &chainSlot,
+		}
+		s := &Server{
+			Stater:                &testutil.MockStater{CustomError: errors.New("internal error")},
+			TimeFetcher:           chain,
+			SyncChecker:           &mockSync.Sync{IsSyncing: false},
+			OptimisticModeFetcher: chain,
+			HeadFetcher:           chain,
+		}
+
+		request := httptest.NewRequest(http.MethodGet, "http://www.example.com/eth/v1/validator/duties/proposer/{epoch}", nil)
+		request.SetPathValue("epoch", "0")
+		writer := httptest.NewRecorder()
+		writer.Body = &bytes.Buffer{}
+
+		s.GetProposerDuties(writer, request)
+		assert.Equal(t, http.StatusInternalServerError, writer.Code)
+		e := &httputil.DefaultJsonError{}
+		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
+		assert.Equal(t, http.StatusInternalServerError, e.Code)
+	})
 }
 
 func TestGetSyncCommitteeDuties(t *testing.T) {
@@ -2457,7 +2570,7 @@ func TestGetSyncCommitteeDuties(t *testing.T) {
 	}
 	require.NoError(t, st.SetNextSyncCommittee(nextCommittee))
 
-	mockChainService := &mockChain.ChainService{Genesis: genesisTime}
+	mockChainService := &mockChain.ChainService{Genesis: genesisTime, State: st}
 	s := &Server{
 		Stater:                &testutil.MockStater{BeaconState: st},
 		SyncChecker:           &mockSync.Sync{IsSyncing: false},
@@ -2648,7 +2761,7 @@ func TestGetSyncCommitteeDuties(t *testing.T) {
 				return newSyncPeriodSt
 			}
 		}
-		mockChainService := &mockChain.ChainService{Genesis: genesisTime, Slot: &newSyncPeriodStartSlot}
+		mockChainService := &mockChain.ChainService{Genesis: genesisTime, Slot: &newSyncPeriodStartSlot, State: newSyncPeriodSt}
 		s := &Server{
 			Stater:                &testutil.MockStater{BeaconState: stateFetchFn(newSyncPeriodStartSlot)},
 			SyncChecker:           &mockSync.Sync{IsSyncing: false},
@@ -2729,8 +2842,7 @@ func TestGetSyncCommitteeDuties(t *testing.T) {
 		slot, err := slots.EpochStart(1)
 		require.NoError(t, err)
 
-		st2, err := util.NewBeaconStateBellatrix()
-		require.NoError(t, err)
+		st2 := st.Copy()
 		require.NoError(t, st2.SetSlot(slot))
 
 		mockChainService := &mockChain.ChainService{
@@ -2744,7 +2856,7 @@ func TestGetSyncCommitteeDuties(t *testing.T) {
 			State: st2,
 		}
 		s := &Server{
-			Stater:                &testutil.MockStater{BeaconState: st},
+			Stater:                &testutil.MockStater{BeaconState: st2},
 			SyncChecker:           &mockSync.Sync{IsSyncing: false},
 			TimeFetcher:           mockChainService,
 			HeadFetcher:           mockChainService,
@@ -2789,6 +2901,62 @@ func TestGetSyncCommitteeDuties(t *testing.T) {
 		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
 		assert.Equal(t, http.StatusServiceUnavailable, e.Code)
 	})
+	t.Run("state not found returns 404", func(t *testing.T) {
+		slot := 2 * params.BeaconConfig().SlotsPerEpoch
+		chainService := &mockChain.ChainService{
+			Slot: &slot,
+		}
+		stateNotFoundErr := lookup.NewStateNotFoundError(8192, []byte("test"))
+		s := &Server{
+			Stater:                &testutil.MockStater{CustomError: &stateNotFoundErr},
+			TimeFetcher:           chainService,
+			SyncChecker:           &mockSync.Sync{IsSyncing: false},
+			OptimisticModeFetcher: chainService,
+			HeadFetcher:           chainService,
+		}
+
+		var body bytes.Buffer
+		_, err := body.WriteString("[\"1\"]")
+		require.NoError(t, err)
+		request := httptest.NewRequest(http.MethodGet, "http://www.example.com/eth/v1/validator/duties/sync/{epoch}", &body)
+		request.SetPathValue("epoch", "1")
+		writer := httptest.NewRecorder()
+		writer.Body = &bytes.Buffer{}
+
+		s.GetSyncCommitteeDuties(writer, request)
+		assert.Equal(t, http.StatusNotFound, writer.Code)
+		e := &httputil.DefaultJsonError{}
+		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
+		assert.Equal(t, http.StatusNotFound, e.Code)
+		assert.StringContains(t, "State not found", e.Message)
+	})
+	t.Run("state fetch error returns 500", func(t *testing.T) {
+		slot := 2 * params.BeaconConfig().SlotsPerEpoch
+		chainService := &mockChain.ChainService{
+			Slot: &slot,
+		}
+		s := &Server{
+			Stater:                &testutil.MockStater{CustomError: errors.New("internal error")},
+			TimeFetcher:           chainService,
+			SyncChecker:           &mockSync.Sync{IsSyncing: false},
+			OptimisticModeFetcher: chainService,
+			HeadFetcher:           chainService,
+		}
+
+		var body bytes.Buffer
+		_, err := body.WriteString("[\"1\"]")
+		require.NoError(t, err)
+		request := httptest.NewRequest(http.MethodGet, "http://www.example.com/eth/v1/validator/duties/sync/{epoch}", &body)
+		request.SetPathValue("epoch", "1")
+		writer := httptest.NewRecorder()
+		writer.Body = &bytes.Buffer{}
+
+		s.GetSyncCommitteeDuties(writer, request)
+		assert.Equal(t, http.StatusInternalServerError, writer.Code)
+		e := &httputil.DefaultJsonError{}
+		require.NoError(t, json.Unmarshal(writer.Body.Bytes(), e))
+		assert.Equal(t, http.StatusInternalServerError, e.Code)
+	})
 }
 
 func TestPrepareBeaconProposer(t *testing.T) {

beacon-chain/rpc/lookup/BUILD.bazel

@@ -11,6 +11,7 @@ go_library(
     deps = [
         "//beacon-chain/blockchain:go_default_library",
         "//beacon-chain/core/peerdas:go_default_library",
+        "//beacon-chain/core/transition:go_default_library",
         "//beacon-chain/db:go_default_library",
         "//beacon-chain/db/filesystem:go_default_library",
         "//beacon-chain/rpc/core:go_default_library",

beacon-chain/rpc/lookup/stater.go

@@ -8,6 +8,7 @@ import (
 	"strings"
 
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/blockchain"
+	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/transition"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/db"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/state"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/state/stategen"
@@ -98,6 +99,7 @@ type Stater interface {
 	State(ctx context.Context, id []byte) (state.BeaconState, error)
 	StateRoot(ctx context.Context, id []byte) ([]byte, error)
 	StateBySlot(ctx context.Context, slot primitives.Slot) (state.BeaconState, error)
+	StateByEpoch(ctx context.Context, epoch primitives.Epoch) (state.BeaconState, error)
 }
 
 // BeaconDbStater is an implementation of Stater. It retrieves states from the beacon chain database.
@@ -267,6 +269,46 @@ func (p *BeaconDbStater) StateBySlot(ctx context.Context, target primitives.Slot
 	return st, nil
 }
 
+// StateByEpoch returns the state for the start of the requested epoch.
+// For current or next epoch, it uses the head state and next slot cache for efficiency.
+// For past epochs, it replays blocks from the most recent canonical state.
+func (p *BeaconDbStater) StateByEpoch(ctx context.Context, epoch primitives.Epoch) (state.BeaconState, error) {
+	ctx, span := trace.StartSpan(ctx, "statefetcher.StateByEpoch")
+	defer span.End()
+
+	targetSlot, err := slots.EpochStart(epoch)
+	if err != nil {
+		return nil, errors.Wrap(err, "could not get epoch start slot")
+	}
+
+	currentSlot := p.GenesisTimeFetcher.CurrentSlot()
+	currentEpoch := slots.ToEpoch(currentSlot)
+
+	// For past epochs, use the replay mechanism
+	if epoch < currentEpoch {
+		return p.StateBySlot(ctx, targetSlot)
+	}
+
+	// For current or next epoch, use head state + next slot cache (much faster)
+	headState, err := p.ChainInfoFetcher.HeadState(ctx)
+	if err != nil {
+		return nil, errors.Wrap(err, "could not get head state")
+	}
+
+	// If head state is already at or past the target slot, return it
+	if headState.Slot() >= targetSlot {
+		return headState, nil
+	}
+
+	// Process slots using the next slot cache
+	headRoot := p.ChainInfoFetcher.CachedHeadRoot()
+	st, err := transition.ProcessSlotsUsingNextSlotCache(ctx, headState, headRoot[:], targetSlot)
+	if err != nil {
+		return nil, errors.Wrapf(err, "could not process slots up to %d", targetSlot)
+	}
+	return st, nil
+}
+
 func (p *BeaconDbStater) headStateRoot(ctx context.Context) ([]byte, error) {
 	b, err := p.ChainInfoFetcher.HeadBlock(ctx)
 	if err != nil {

beacon-chain/rpc/lookup/stater_test.go

@@ -444,3 +444,111 @@ func TestStateBySlot_AfterHeadSlot(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, primitives.Slot(101), st.Slot())
 }
+
+func TestStateByEpoch(t *testing.T) {
+	ctx := t.Context()
+	slotsPerEpoch := params.BeaconConfig().SlotsPerEpoch
+
+	t.Run("current epoch uses head state", func(t *testing.T) {
+		// Head is at slot 5 (epoch 0), requesting epoch 0
+		headSlot := primitives.Slot(5)
+		headSt, err := statenative.InitializeFromProtoPhase0(&ethpb.BeaconState{Slot: headSlot})
+		require.NoError(t, err)
+
+		currentSlot := headSlot
+		mock := &chainMock.ChainService{State: headSt, Slot: &currentSlot}
+		p := BeaconDbStater{ChainInfoFetcher: mock, GenesisTimeFetcher: mock}
+
+		st, err := p.StateByEpoch(ctx, 0)
+		require.NoError(t, err)
+		// Should return head state since it's already past epoch start
+		assert.Equal(t, headSlot, st.Slot())
+	})
+
+	t.Run("current epoch processes slots to epoch start", func(t *testing.T) {
+		// Head is at slot 5 (epoch 0), requesting epoch 1
+		// Current slot is 32 (epoch 1), so epoch 1 is current epoch
+		headSlot := primitives.Slot(5)
+		headSt, err := statenative.InitializeFromProtoPhase0(&ethpb.BeaconState{Slot: headSlot})
+		require.NoError(t, err)
+
+		currentSlot := slotsPerEpoch // slot 32, epoch 1
+		mock := &chainMock.ChainService{State: headSt, Slot: &currentSlot}
+		p := BeaconDbStater{ChainInfoFetcher: mock, GenesisTimeFetcher: mock}
+
+		// Note: This will fail since ProcessSlotsUsingNextSlotCache requires proper setup
+		// In real usage, the transition package handles this properly
+		_, err = p.StateByEpoch(ctx, 1)
+		// The error is expected since we don't have a fully initialized beacon state
+		// that can process slots (missing committees, etc.)
+		assert.NotNil(t, err)
+	})
+
+	t.Run("past epoch uses replay", func(t *testing.T) {
+		// Head is at epoch 2, requesting epoch 0 (past)
+		headSlot := slotsPerEpoch * 2 // slot 64, epoch 2
+		headSt, err := statenative.InitializeFromProtoPhase0(&ethpb.BeaconState{Slot: headSlot})
+		require.NoError(t, err)
+
+		pastEpochSt, err := statenative.InitializeFromProtoPhase0(&ethpb.BeaconState{Slot: 0})
+		require.NoError(t, err)
+
+		currentSlot := headSlot
+		mock := &chainMock.ChainService{State: headSt, Slot: &currentSlot}
+		mockReplayer := mockstategen.NewReplayerBuilder()
+		mockReplayer.SetMockStateForSlot(pastEpochSt, 0)
+		p := BeaconDbStater{ChainInfoFetcher: mock, GenesisTimeFetcher: mock, ReplayerBuilder: mockReplayer}
+
+		st, err := p.StateByEpoch(ctx, 0)
+		require.NoError(t, err)
+		assert.Equal(t, primitives.Slot(0), st.Slot())
+	})
+
+	t.Run("next epoch uses head state path", func(t *testing.T) {
+		// Head is at slot 30 (epoch 0), requesting epoch 1 (next)
+		// Current slot is 30 (epoch 0), so epoch 1 is next epoch
+		headSlot := primitives.Slot(30)
+		headSt, err := statenative.InitializeFromProtoPhase0(&ethpb.BeaconState{Slot: headSlot})
+		require.NoError(t, err)
+
+		currentSlot := headSlot
+		mock := &chainMock.ChainService{State: headSt, Slot: &currentSlot}
+		p := BeaconDbStater{ChainInfoFetcher: mock, GenesisTimeFetcher: mock}
+
+		// Note: This will fail since ProcessSlotsUsingNextSlotCache requires proper setup
+		_, err = p.StateByEpoch(ctx, 1)
+		// The error is expected since we don't have a fully initialized beacon state
+		assert.NotNil(t, err)
+	})
+
+	t.Run("head state already at target slot returns immediately", func(t *testing.T) {
+		// Head is at slot 32 (epoch 1 start), requesting epoch 1
+		headSlot := slotsPerEpoch // slot 32
+		headSt, err := statenative.InitializeFromProtoPhase0(&ethpb.BeaconState{Slot: headSlot})
+		require.NoError(t, err)
+
+		currentSlot := headSlot
+		mock := &chainMock.ChainService{State: headSt, Slot: &currentSlot}
+		p := BeaconDbStater{ChainInfoFetcher: mock, GenesisTimeFetcher: mock}
+
+		st, err := p.StateByEpoch(ctx, 1)
+		require.NoError(t, err)
+		assert.Equal(t, headSlot, st.Slot())
+	})
+
+	t.Run("head state past target slot returns head state", func(t *testing.T) {
+		// Head is at slot 40, requesting epoch 1 (starts at slot 32)
+		headSlot := primitives.Slot(40)
+		headSt, err := statenative.InitializeFromProtoPhase0(&ethpb.BeaconState{Slot: headSlot})
+		require.NoError(t, err)
+
+		currentSlot := headSlot
+		mock := &chainMock.ChainService{State: headSt, Slot: &currentSlot}
+		p := BeaconDbStater{ChainInfoFetcher: mock, GenesisTimeFetcher: mock}
+
+		st, err := p.StateByEpoch(ctx, 1)
+		require.NoError(t, err)
+		// Returns head state since it's already >= epoch start
+		assert.Equal(t, headSlot, st.Slot())
+	})
+}

beacon-chain/rpc/prysm/v1alpha1/validator/attester.go

@@ -52,24 +52,27 @@ func (vs *Server) ProposeAttestation(ctx context.Context, att *ethpb.Attestation
 	ctx, span := trace.StartSpan(ctx, "AttesterServer.ProposeAttestation")
 	defer span.End()
 
+	if vs.SyncChecker.Syncing() {
+		return nil, status.Errorf(codes.Unavailable, "Syncing to latest head, not ready to respond")
+	}
+
 	resp, err := vs.proposeAtt(ctx, att, att.GetData().CommitteeIndex)
 	if err != nil {
 		return nil, err
 	}
 
-	if features.Get().EnableExperimentalAttestationPool {
-		if err = vs.AttestationCache.Add(att); err != nil {
-			log.WithError(err).Error("Could not save attestation")
-		}
-	} else {
-		go func() {
+	go func() {
+		if features.Get().EnableExperimentalAttestationPool {
+			if err := vs.AttestationCache.Add(att); err != nil {
+				log.WithError(err).Error("Could not save attestation")
+			}
+		} else {
 			attCopy := att.Copy()
 			if err := vs.AttPool.SaveUnaggregatedAttestation(attCopy); err != nil {
 				log.WithError(err).Error("Could not save unaggregated attestation")
-				return
 			}
-		}()
-	}
+		}
+	}()
 
 	return resp, nil
 }
@@ -82,6 +85,10 @@ func (vs *Server) ProposeAttestationElectra(ctx context.Context, singleAtt *ethp
 	ctx, span := trace.StartSpan(ctx, "AttesterServer.ProposeAttestationElectra")
 	defer span.End()
 
+	if vs.SyncChecker.Syncing() {
+		return nil, status.Errorf(codes.Unavailable, "Syncing to latest head, not ready to respond")
+	}
+
 	resp, err := vs.proposeAtt(ctx, singleAtt, singleAtt.GetCommitteeIndex())
 	if err != nil {
 		return nil, err
@@ -98,18 +105,17 @@ func (vs *Server) ProposeAttestationElectra(ctx context.Context, singleAtt *ethp
 
 	singleAttCopy := singleAtt.Copy()
 	att := singleAttCopy.ToAttestationElectra(committee)
-	if features.Get().EnableExperimentalAttestationPool {
-		if err = vs.AttestationCache.Add(att); err != nil {
-			log.WithError(err).Error("Could not save attestation")
-		}
-	} else {
-		go func() {
+	go func() {
+		if features.Get().EnableExperimentalAttestationPool {
+			if err := vs.AttestationCache.Add(att); err != nil {
+				log.WithError(err).Error("Could not save attestation")
+			}
+		} else {
 			if err := vs.AttPool.SaveUnaggregatedAttestation(att); err != nil {
 				log.WithError(err).Error("Could not save unaggregated attestation")
-				return
 			}
-		}()
-	}
+		}
+	}()
 
 	return resp, nil
 }

beacon-chain/rpc/prysm/v1alpha1/validator/attester_test.go

@@ -38,6 +38,7 @@ func TestProposeAttestation(t *testing.T) {
 		OperationNotifier:       (&mock.ChainService{}).OperationNotifier(),
 		TimeFetcher:             chainService,
 		AttestationStateFetcher: chainService,
+		SyncChecker:             &mockSync.Sync{IsSyncing: false},
 	}
 	head := util.NewBeaconBlock()
 	head.Block.Slot = 999
@@ -141,6 +142,7 @@ func TestProposeAttestation_IncorrectSignature(t *testing.T) {
 		P2P:               &mockp2p.MockBroadcaster{},
 		AttPool:           attestations.NewPool(),
 		OperationNotifier: (&mock.ChainService{}).OperationNotifier(),
+		SyncChecker:       &mockSync.Sync{IsSyncing: false},
 	}
 
 	req := util.HydrateAttestation(&ethpb.Attestation{})
@@ -149,6 +151,37 @@ func TestProposeAttestation_IncorrectSignature(t *testing.T) {
 	assert.ErrorContains(t, wanted, err)
 }
 
+func TestProposeAttestation_Syncing(t *testing.T) {
+	attesterServer := &Server{
+		SyncChecker: &mockSync.Sync{IsSyncing: true},
+	}
+
+	req := util.HydrateAttestation(&ethpb.Attestation{})
+	_, err := attesterServer.ProposeAttestation(t.Context(), req)
+	assert.ErrorContains(t, "Syncing to latest head", err)
+	s, ok := status.FromError(err)
+	require.Equal(t, true, ok)
+	assert.Equal(t, codes.Unavailable, s.Code())
+}
+
+func TestProposeAttestationElectra_Syncing(t *testing.T) {
+	attesterServer := &Server{
+		SyncChecker: &mockSync.Sync{IsSyncing: true},
+	}
+
+	req := &ethpb.SingleAttestation{
+		Data: &ethpb.AttestationData{
+			Source: &ethpb.Checkpoint{Root: make([]byte, 32)},
+			Target: &ethpb.Checkpoint{Root: make([]byte, 32)},
+		},
+	}
+	_, err := attesterServer.ProposeAttestationElectra(t.Context(), req)
+	assert.ErrorContains(t, "Syncing to latest head", err)
+	s, ok := status.FromError(err)
+	require.Equal(t, true, ok)
+	assert.Equal(t, codes.Unavailable, s.Code())
+}
+
 func TestGetAttestationData_OK(t *testing.T) {
 	block := util.NewBeaconBlock()
 	block.Block.Slot = 3*params.BeaconConfig().SlotsPerEpoch + 1

beacon-chain/rpc/testutil/BUILD.bazel

@@ -26,5 +26,6 @@ go_library(
         "//proto/prysm/v1alpha1:go_default_library",
         "//testing/require:go_default_library",
         "//testing/util:go_default_library",
+        "//time/slots:go_default_library",
     ],
 )

beacon-chain/rpc/testutil/mock_stater.go

@@ -6,6 +6,7 @@ import (
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/state"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/primitives"
 	"github.com/OffchainLabs/prysm/v7/encoding/bytesutil"
+	"github.com/OffchainLabs/prysm/v7/time/slots"
 )
 
 // MockStater is a fake implementation of lookup.Stater.
@@ -14,6 +15,7 @@ type MockStater struct {
 	StateProviderFunc func(ctx context.Context, stateId []byte) (state.BeaconState, error)
 	BeaconStateRoot   []byte
 	StatesBySlot      map[primitives.Slot]state.BeaconState
+	StatesByEpoch     map[primitives.Epoch]state.BeaconState
 	StatesByRoot      map[[32]byte]state.BeaconState
 	CustomError       error
 }
@@ -43,3 +45,22 @@ func (m *MockStater) StateRoot(context.Context, []byte) ([]byte, error) {
 func (m *MockStater) StateBySlot(_ context.Context, s primitives.Slot) (state.BeaconState, error) {
 	return m.StatesBySlot[s], nil
 }
+
+// StateByEpoch --
+func (m *MockStater) StateByEpoch(_ context.Context, e primitives.Epoch) (state.BeaconState, error) {
+	if m.CustomError != nil {
+		return nil, m.CustomError
+	}
+	if m.StatesByEpoch != nil {
+		return m.StatesByEpoch[e], nil
+	}
+	// Fall back to StatesBySlot if StatesByEpoch is not set
+	slot, err := slots.EpochStart(e)
+	if err != nil {
+		return nil, err
+	}
+	if m.StatesBySlot != nil {
+		return m.StatesBySlot[slot], nil
+	}
+	return m.BeaconState, nil
+}

beacon-chain/sync/batch_verifier.go

@@ -161,13 +161,17 @@ func (s *Service) validateWithKzgBatchVerifier(ctx context.Context, dataColumns
 
 	timeout := time.Duration(params.BeaconConfig().SecondsPerSlot) * time.Second
 
-	resChan := make(chan error)
+	resChan := make(chan error, 1)
 	verificationSet := &kzgVerifier{dataColumns: dataColumns, resChan: resChan}
-	s.kzgChan <- verificationSet
-
 	ctx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
 
+	select {
+	case s.kzgChan <- verificationSet:
+	case <-ctx.Done():
+		return pubsub.ValidationIgnore, ctx.Err()
+	}
+
 	select {
 	case <-ctx.Done():
 		return pubsub.ValidationIgnore, ctx.Err() // parent context canceled, give up

beacon-chain/sync/data_column_sidecars.go

@@ -3,6 +3,7 @@ package sync
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"maps"
 	"slices"
 	"sync"
@@ -243,8 +244,10 @@ func requestDirectSidecarsFromPeers(
 	}
 
 	// Compute missing indices by root, excluding those already in storage.
+	var lastRoot [fieldparams.RootLength]byte
 	missingIndicesByRoot := make(map[[fieldparams.RootLength]byte]map[uint64]bool, len(incompleteRoots))
 	for root := range incompleteRoots {
+		lastRoot = root
 		storedIndices := storedIndicesByRoot[root]
 
 		missingIndices := make(map[uint64]bool, len(requestedIndices))
@@ -259,6 +262,7 @@ func requestDirectSidecarsFromPeers(
 		}
 	}
 
+	initialMissingRootCount := len(missingIndicesByRoot)
 	initialMissingCount := computeTotalCount(missingIndicesByRoot)
 
 	indicesByRootByPeer, err := computeIndicesByRootByPeer(params.P2P, slotByRoot, missingIndicesByRoot, connectedPeers)
@@ -301,11 +305,19 @@ func requestDirectSidecarsFromPeers(
 		}
 	}
 
-	log.WithFields(logrus.Fields{
-		"duration":            time.Since(start),
-		"initialMissingCount": initialMissingCount,
-		"finalMissingCount":   computeTotalCount(missingIndicesByRoot),
-	}).Debug("Requested direct data column sidecars from peers")
+	log := log.WithFields(logrus.Fields{
+		"duration":                time.Since(start),
+		"initialMissingRootCount": initialMissingRootCount,
+		"initialMissingCount":     initialMissingCount,
+		"finalMissingRootCount":   len(missingIndicesByRoot),
+		"finalMissingCount":       computeTotalCount(missingIndicesByRoot),
+	})
+
+	if initialMissingRootCount == 1 {
+		log = log.WithField("root", fmt.Sprintf("%#x", lastRoot))
+	}
+
+	log.Debug("Requested direct data column sidecars from peers")
 
 	return verifiedColumnsByRoot, nil
 }

beacon-chain/sync/kzg_batch_verifier_test.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/blockchain/kzg"
+	"github.com/OffchainLabs/prysm/v7/config/params"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/blocks"
 	ethpb "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
 	"github.com/OffchainLabs/prysm/v7/testing/assert"
@@ -268,6 +269,71 @@ func TestKzgBatchVerifierFallback(t *testing.T) {
 	})
 }
 
+func TestValidateWithKzgBatchVerifier_DeadlockOnTimeout(t *testing.T) {
+	err := kzg.Start()
+	require.NoError(t, err)
+
+	params.SetupTestConfigCleanup(t)
+	cfg := params.BeaconConfig().Copy()
+	cfg.SecondsPerSlot = 0
+	params.OverrideBeaconConfig(cfg)
+
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+
+	service := &Service{
+		ctx:     ctx,
+		kzgChan: make(chan *kzgVerifier),
+	}
+	go service.kzgVerifierRoutine()
+
+	result, err := service.validateWithKzgBatchVerifier(context.Background(), nil)
+	require.Equal(t, pubsub.ValidationIgnore, result)
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+
+	done := make(chan struct{})
+	go func() {
+		_, _ = service.validateWithKzgBatchVerifier(context.Background(), nil)
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("validateWithKzgBatchVerifier blocked")
+	}
+}
+
+func TestValidateWithKzgBatchVerifier_ContextCanceledBeforeSend(t *testing.T) {
+	cancelledCtx, cancel := context.WithCancel(t.Context())
+	cancel()
+
+	service := &Service{
+		ctx:     context.Background(),
+		kzgChan: make(chan *kzgVerifier),
+	}
+
+	done := make(chan struct{})
+	go func() {
+		result, err := service.validateWithKzgBatchVerifier(cancelledCtx, nil)
+		require.Equal(t, pubsub.ValidationIgnore, result)
+		require.ErrorIs(t, err, context.Canceled)
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(500 * time.Millisecond):
+		t.Fatal("validateWithKzgBatchVerifier did not return after context cancellation")
+	}
+
+	select {
+	case <-service.kzgChan:
+		t.Fatal("verificationSet was sent to kzgChan despite canceled context")
+	default:
+	}
+}
+
 func createValidTestDataColumns(t *testing.T, count int) []blocks.RODataColumn {
 	_, roSidecars, _ := util.GenerateTestFuluBlockWithSidecars(t, count)
 	if len(roSidecars) >= count {

beacon-chain/sync/metrics.go

@@ -204,6 +204,13 @@ var (
 		},
 	)
 
+	dataColumnsRecoveredFromELAttempts = promauto.NewCounter(
+		prometheus.CounterOpts{
+			Name: "data_columns_recovered_from_el_attempts",
+			Help: "Count the number of data columns recovery attempts from the execution layer.",
+		},
+	)
+
 	dataColumnsRecoveredFromELTotal = promauto.NewCounter(
 		prometheus.CounterOpts{
 			Name: "data_columns_recovered_from_el_total",
@@ -242,6 +249,13 @@ var (
 			Buckets: []float64{100, 250, 500, 750, 1000, 1500, 2000, 4000, 8000, 12000, 16000},
 		},
 	)
+
+	dataColumnSidecarsObtainedViaELCount = promauto.NewSummary(
+		prometheus.SummaryOpts{
+			Name: "data_column_obtained_via_el_count",
+			Help: "Count the number of data column sidecars obtained via the execution layer.",
+		},
+	)
 )
 
 func (s *Service) updateMetrics() {

beacon-chain/sync/pending_attestations_queue.go

@@ -3,9 +3,9 @@ package sync
 import (
 	"bytes"
 	"context"
-	"encoding/hex"
 	"fmt"
 	"slices"
+	"time"
 
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/blockchain"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/blocks"
@@ -21,13 +21,23 @@ import (
 	"github.com/OffchainLabs/prysm/v7/monitoring/tracing/trace"
 	ethpb "github.com/OffchainLabs/prysm/v7/proto/prysm/v1alpha1"
 	"github.com/OffchainLabs/prysm/v7/runtime/version"
-	"github.com/OffchainLabs/prysm/v7/time"
 	"github.com/OffchainLabs/prysm/v7/time/slots"
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
+	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 )
 
-var pendingAttsLimit = 32768
+const pendingAttsLimit = 32768
+
+// aggregatorIndexFilter defines how aggregator index should be handled in equality checks.
+type aggregatorIndexFilter int
+
+const (
+	// ignoreAggregatorIndex means aggregates differing only by aggregator index are considered equal.
+	ignoreAggregatorIndex aggregatorIndexFilter = iota
+	// includeAggregatorIndex means aggregator index must also match for aggregates to be considered equal.
+	includeAggregatorIndex
+)
 
 // This method processes pending attestations as a "known" block as arrived. With validations,
 // the valid attestations get saved into the operation mem pool, and the invalid attestations gets deleted
@@ -50,16 +60,7 @@ func (s *Service) processPendingAttsForBlock(ctx context.Context, bRoot [32]byte
 	attestations := s.blkRootToPendingAtts[bRoot]
 	s.pendingAttsLock.RUnlock()
 
-	if len(attestations) > 0 {
-		start := time.Now()
-		s.processAttestations(ctx, attestations)
-		duration := time.Since(start)
-		log.WithFields(logrus.Fields{
-			"blockRoot":        hex.EncodeToString(bytesutil.Trunc(bRoot[:])),
-			"pendingAttsCount": len(attestations),
-			"duration":         duration,
-		}).Debug("Verified and saved pending attestations to pool")
-	}
+	s.processAttestations(ctx, attestations)
 
 	randGen := rand.NewGenerator()
 	// Delete the missing block root key from pending attestation queue so a node will not request for the block again.
@@ -79,26 +80,71 @@ func (s *Service) processPendingAttsForBlock(ctx context.Context, bRoot [32]byte
 	return s.sendBatchRootRequest(ctx, pendingRoots, randGen)
 }
 
+// processAttestations processes a list of attestations.
+// It assumes (for logging purposes only) that all attestations pertain to the same block.
 func (s *Service) processAttestations(ctx context.Context, attestations []any) {
 	if len(attestations) == 0 {
 		return
 	}
 
+	firstAttestation := attestations[0]
+	var blockRoot []byte
+	switch v := firstAttestation.(type) {
+	case ethpb.Att:
+		blockRoot = v.GetData().BeaconBlockRoot
+	case ethpb.SignedAggregateAttAndProof:
+		blockRoot = v.AggregateAttestationAndProof().AggregateVal().GetData().BeaconBlockRoot
+	default:
+		log.Warnf("Unexpected attestation type %T, skipping processing", v)
+		return
+	}
+
+	validAggregates := make([]ethpb.SignedAggregateAttAndProof, 0, len(attestations))
+	startAggregate := time.Now()
 	atts := make([]ethpb.Att, 0, len(attestations))
+	aggregateAttAndProofCount := 0
 	for _, att := range attestations {
 		switch v := att.(type) {
 		case ethpb.Att:
 			atts = append(atts, v)
 		case ethpb.SignedAggregateAttAndProof:
-			s.processAggregate(ctx, v)
+			aggregateAttAndProofCount++
+			// Avoid processing multiple aggregates only differing by aggregator index.
+			if slices.ContainsFunc(validAggregates, func(other ethpb.SignedAggregateAttAndProof) bool {
+				return pendingAggregatesAreEqual(v, other, ignoreAggregatorIndex)
+			}) {
+				continue
+			}
+
+			if err := s.processAggregate(ctx, v); err != nil {
+				log.WithError(err).Debug("Pending aggregate attestation could not be processed")
+				continue
+			}
+
+			validAggregates = append(validAggregates, v)
 		default:
 			log.Warnf("Unexpected attestation type %T, skipping", v)
 		}
 	}
+	durationAggregateAttAndProof := time.Since(startAggregate)
 
+	startAtts := time.Now()
 	for _, bucket := range bucketAttestationsByData(atts) {
 		s.processAttestationBucket(ctx, bucket)
 	}
+
+	durationAtts := time.Since(startAtts)
+
+	log.WithFields(logrus.Fields{
+		"blockRoot":                       fmt.Sprintf("%#x", blockRoot),
+		"totalCount":                      len(attestations),
+		"aggregateAttAndProofCount":       aggregateAttAndProofCount,
+		"uniqueAggregateAttAndProofCount": len(validAggregates),
+		"attCount":                        len(atts),
+		"durationTotal":                   durationAggregateAttAndProof + durationAtts,
+		"durationAggregateAttAndProof":    durationAggregateAttAndProof,
+		"durationAtts":                    durationAtts,
+	}).Debug("Verified and saved pending attestations to pool")
 }
 
 // attestationBucket groups attestations with the same AttestationData for batch processing.
@@ -303,21 +349,20 @@ func (s *Service) processVerifiedAttestation(
 	})
 }
 
-func (s *Service) processAggregate(ctx context.Context, aggregate ethpb.SignedAggregateAttAndProof) {
+func (s *Service) processAggregate(ctx context.Context, aggregate ethpb.SignedAggregateAttAndProof) error {
 	res, err := s.validateAggregatedAtt(ctx, aggregate)
 	if err != nil {
 		log.WithError(err).Debug("Pending aggregated attestation failed validation")
-		return
+		return errors.Wrap(err, "validate aggregated att")
 	}
+
 	if res != pubsub.ValidationAccept || !s.validateBlockInAttestation(ctx, aggregate) {
-		log.Debug("Pending aggregated attestation failed validation")
-		return
+		return errors.New("Pending aggregated attestation failed validation")
 	}
 
 	att := aggregate.AggregateAttestationAndProof().AggregateVal()
 	if err := s.saveAttestation(att); err != nil {
-		log.WithError(err).Debug("Could not save aggregated attestation")
-		return
+		return errors.Wrap(err, "save attestation")
 	}
 
 	_ = s.setAggregatorIndexEpochSeen(att.GetData().Target.Epoch, aggregate.AggregateAttestationAndProof().GetAggregatorIndex())
@@ -325,6 +370,8 @@ func (s *Service) processAggregate(ctx context.Context, aggregate ethpb.SignedAg
 	if err := s.cfg.p2p.Broadcast(ctx, aggregate); err != nil {
 		log.WithError(err).Debug("Could not broadcast aggregated attestation")
 	}
+
+	return nil
 }
 
 // This defines how pending aggregates are saved in the map. The key is the
@@ -336,7 +383,7 @@ func (s *Service) savePendingAggregate(agg ethpb.SignedAggregateAttAndProof) {
 
 	s.savePending(root, agg, func(other any) bool {
 		a, ok := other.(ethpb.SignedAggregateAttAndProof)
-		return ok && pendingAggregatesAreEqual(agg, a)
+		return ok && pendingAggregatesAreEqual(agg, a, includeAggregatorIndex)
 	})
 }
 
@@ -391,13 +438,19 @@ func (s *Service) savePending(root [32]byte, pending any, isEqual func(other any
 	s.blkRootToPendingAtts[root] = append(s.blkRootToPendingAtts[root], pending)
 }
 
-func pendingAggregatesAreEqual(a, b ethpb.SignedAggregateAttAndProof) bool {
+// pendingAggregatesAreEqual checks if two pending aggregate attestations are equal.
+// The filter parameter controls whether aggregator index is considered in the equality check.
+func pendingAggregatesAreEqual(a, b ethpb.SignedAggregateAttAndProof, filter aggregatorIndexFilter) bool {
 	if a.Version() != b.Version() {
 		return false
 	}
-	if a.AggregateAttestationAndProof().GetAggregatorIndex() != b.AggregateAttestationAndProof().GetAggregatorIndex() {
-		return false
+
+	if filter == includeAggregatorIndex {
+		if a.AggregateAttestationAndProof().GetAggregatorIndex() != b.AggregateAttestationAndProof().GetAggregatorIndex() {
+			return false
+		}
 	}
+
 	aAtt := a.AggregateAttestationAndProof().AggregateVal()
 	bAtt := b.AggregateAttestationAndProof().AggregateVal()
 	if aAtt.GetData().Slot != bAtt.GetData().Slot {

beacon-chain/sync/pending_attestations_queue_test.go

@@ -94,7 +94,7 @@ func TestProcessPendingAtts_NoBlockRequestBlock(t *testing.T) {
 	// Process block A (which exists and has no pending attestations)
 	// This should skip processing attestations for A and request blocks B and C
 	require.NoError(t, r.processPendingAttsForBlock(t.Context(), rootA))
-	require.LogsContain(t, hook, "Requesting block by root")
+	require.LogsContain(t, hook, "Requesting blocks by root")
 }
 
 func TestProcessPendingAtts_HasBlockSaveUnaggregatedAtt(t *testing.T) {
@@ -911,17 +911,17 @@ func Test_pendingAggregatesAreEqual(t *testing.T) {
 					},
 					AggregationBits: bitfield.Bitlist{0b1111},
 				}}}
-		assert.Equal(t, true, pendingAggregatesAreEqual(a, b))
+		assert.Equal(t, true, pendingAggregatesAreEqual(a, b, includeAggregatorIndex))
 	})
 	t.Run("different version", func(t *testing.T) {
 		a := &ethpb.SignedAggregateAttestationAndProof{Message: &ethpb.AggregateAttestationAndProof{AggregatorIndex: 1}}
 		b := &ethpb.SignedAggregateAttestationAndProofElectra{Message: &ethpb.AggregateAttestationAndProofElectra{AggregatorIndex: 1}}
-		assert.Equal(t, false, pendingAggregatesAreEqual(a, b))
+		assert.Equal(t, false, pendingAggregatesAreEqual(a, b, includeAggregatorIndex))
 	})
 	t.Run("different aggregator index", func(t *testing.T) {
 		a := &ethpb.SignedAggregateAttestationAndProof{Message: &ethpb.AggregateAttestationAndProof{AggregatorIndex: 1}}
 		b := &ethpb.SignedAggregateAttestationAndProof{Message: &ethpb.AggregateAttestationAndProof{AggregatorIndex: 2}}
-		assert.Equal(t, false, pendingAggregatesAreEqual(a, b))
+		assert.Equal(t, false, pendingAggregatesAreEqual(a, b, includeAggregatorIndex))
 	})
 	t.Run("different slot", func(t *testing.T) {
 		a := &ethpb.SignedAggregateAttestationAndProof{
@@ -942,7 +942,7 @@ func Test_pendingAggregatesAreEqual(t *testing.T) {
 					},
 					AggregationBits: bitfield.Bitlist{0b1111},
 				}}}
-		assert.Equal(t, false, pendingAggregatesAreEqual(a, b))
+		assert.Equal(t, false, pendingAggregatesAreEqual(a, b, includeAggregatorIndex))
 	})
 	t.Run("different committee index", func(t *testing.T) {
 		a := &ethpb.SignedAggregateAttestationAndProof{
@@ -963,7 +963,7 @@ func Test_pendingAggregatesAreEqual(t *testing.T) {
 					},
 					AggregationBits: bitfield.Bitlist{0b1111},
 				}}}
-		assert.Equal(t, false, pendingAggregatesAreEqual(a, b))
+		assert.Equal(t, false, pendingAggregatesAreEqual(a, b, includeAggregatorIndex))
 	})
 	t.Run("different aggregation bits", func(t *testing.T) {
 		a := &ethpb.SignedAggregateAttestationAndProof{
@@ -984,7 +984,30 @@ func Test_pendingAggregatesAreEqual(t *testing.T) {
 					},
 					AggregationBits: bitfield.Bitlist{0b1000},
 				}}}
-		assert.Equal(t, false, pendingAggregatesAreEqual(a, b))
+		assert.Equal(t, false, pendingAggregatesAreEqual(a, b, includeAggregatorIndex))
+	})
+	t.Run("different aggregator index should be equal while ignoring aggregator index", func(t *testing.T) {
+		a := &ethpb.SignedAggregateAttestationAndProof{
+			Message: &ethpb.AggregateAttestationAndProof{
+				AggregatorIndex: 1,
+				Aggregate: &ethpb.Attestation{
+					Data: &ethpb.AttestationData{
+						Slot:           1,
+						CommitteeIndex: 1,
+					},
+					AggregationBits: bitfield.Bitlist{0b1111},
+				}}}
+		b := &ethpb.SignedAggregateAttestationAndProof{
+			Message: &ethpb.AggregateAttestationAndProof{
+				AggregatorIndex: 2,
+				Aggregate: &ethpb.Attestation{
+					Data: &ethpb.AttestationData{
+						Slot:           1,
+						CommitteeIndex: 1,
+					},
+					AggregationBits: bitfield.Bitlist{0b1111},
+				}}}
+		assert.Equal(t, true, pendingAggregatesAreEqual(a, b, ignoreAggregatorIndex))
 	})
 }
 

beacon-chain/sync/pending_blocks_queue.go

@@ -2,7 +2,6 @@ package sync
 
 import (
 	"context"
-	"encoding/hex"
 	"fmt"
 	"slices"
 	"sync"
@@ -44,11 +43,13 @@ func (s *Service) processPendingBlocksQueue() {
 		if !s.chainIsStarted() {
 			return
 		}
+
 		locker.Lock()
+		defer locker.Unlock()
+
 		if err := s.processPendingBlocks(s.ctx); err != nil {
 			log.WithError(err).Debug("Could not process pending blocks")
 		}
-		locker.Unlock()
 	})
 }
 
@@ -73,8 +74,10 @@ func (s *Service) processPendingBlocks(ctx context.Context) error {
 	randGen := rand.NewGenerator()
 	var parentRoots [][32]byte
 
+	blkRoots := make([][32]byte, 0, len(sortedSlots)*maxBlocksPerSlot)
+
 	// Iterate through sorted slots.
-	for _, slot := range sortedSlots {
+	for i, slot := range sortedSlots {
 		// Skip processing if slot is in the future.
 		if slot > s.cfg.clock.CurrentSlot() {
 			continue
@@ -91,6 +94,9 @@ func (s *Service) processPendingBlocks(ctx context.Context) error {
 
 		// Process each block in the queue.
 		for _, b := range blocksInCache {
+			start := time.Now()
+			totalDuration := time.Duration(0)
+
 			if err := blocks.BeaconBlockIsNil(b); err != nil {
 				continue
 			}
@@ -147,19 +153,34 @@ func (s *Service) processPendingBlocks(ctx context.Context) error {
 			}
 			cancelFunction()
 
-			// Process pending attestations for this block.
-			if err := s.processPendingAttsForBlock(ctx, blkRoot); err != nil {
-				log.WithError(err).Debug("Failed to process pending attestations for block")
-			}
+			blkRoots = append(blkRoots, blkRoot)
 
 			// Remove the processed block from the queue.
 			if err := s.removeBlockFromQueue(b, blkRoot); err != nil {
 				return err
 			}
-			log.WithFields(logrus.Fields{"slot": slot, "blockRoot": hex.EncodeToString(bytesutil.Trunc(blkRoot[:]))}).Debug("Processed pending block and cleared it in cache")
+
+			duration := time.Since(start)
+			totalDuration += duration
+			log.WithFields(logrus.Fields{
+				"slotIndex":     fmt.Sprintf("%d/%d", i+1, len(sortedSlots)),
+				"slot":          slot,
+				"root":          fmt.Sprintf("%#x", blkRoot),
+				"duration":      duration,
+				"totalDuration": totalDuration,
+			}).Debug("Processed pending block and cleared it in cache")
 		}
+
 		span.End()
 	}
+
+	for _, blkRoot := range blkRoots {
+		// Process pending attestations for this block.
+		if err := s.processPendingAttsForBlock(ctx, blkRoot); err != nil {
+			log.WithError(err).Debug("Failed to process pending attestations for block")
+		}
+	}
+
 	return s.sendBatchRootRequest(ctx, parentRoots, randGen)
 }
 
@@ -379,6 +400,19 @@ func (s *Service) sendBatchRootRequest(ctx context.Context, roots [][32]byte, ra
 			req = roots[:maxReqBlock]
 		}
 
+		if logrus.GetLevel() >= logrus.DebugLevel {
+			rootsStr := make([]string, 0, len(roots))
+			for _, req := range roots {
+				rootsStr = append(rootsStr, fmt.Sprintf("%#x", req))
+			}
+
+			log.WithFields(logrus.Fields{
+				"peer":  pid,
+				"count": len(req),
+				"roots": rootsStr,
+			}).Debug("Requesting blocks by root")
+		}
+
 		// Send the request to the peer.
 		if err := s.sendBeaconBlocksRequest(ctx, &req, pid); err != nil {
 			tracing.AnnotateError(span, err)
@@ -438,8 +472,6 @@ func (s *Service) filterOutPendingAndSynced(roots [][fieldparams.RootLength]byte
 			roots = append(roots[:i], roots[i+1:]...)
 			continue
 		}
-
-		log.WithField("blockRoot", fmt.Sprintf("%#x", r)).Debug("Requesting block by root")
 	}
 	return roots
 }

beacon-chain/sync/subscriber_beacon_blocks.go

@@ -189,12 +189,30 @@ func (s *Service) processDataColumnSidecarsFromExecution(ctx context.Context, so
 		ctx, cancel := context.WithTimeout(ctx, secondsPerHalfSlot)
 		defer cancel()
 
+		log := log.WithFields(logrus.Fields{
+			"root":          fmt.Sprintf("%#x", source.Root()),
+			"slot":          source.Slot(),
+			"proposerIndex": source.ProposerIndex(),
+			"type":          source.Type(),
+		})
+
+		var constructedSidecarCount uint64
 		for iteration := uint64(0); ; /*no stop condition*/ iteration++ {
+			log = log.WithField("iteration", iteration)
+
 			// Exit early if all sidecars to sample have been seen.
 			if s.haveAllSidecarsBeenSeen(source.Slot(), source.ProposerIndex(), columnIndicesToSample) {
+				if iteration > 0 && constructedSidecarCount == 0 {
+					log.Debug("No data column sidecars constructed from the execution client")
+				}
+
 				return nil, nil
 			}
 
+			if iteration == 0 {
+				dataColumnsRecoveredFromELAttempts.Inc()
+			}
+
 			// Try to reconstruct data column constructedSidecars from the execution client.
 			constructedSidecars, err := s.cfg.executionReconstructor.ConstructDataColumnSidecars(ctx, source)
 			if err != nil {
@@ -202,8 +220,8 @@ func (s *Service) processDataColumnSidecarsFromExecution(ctx context.Context, so
 			}
 
 			// No sidecars are retrieved from the EL, retry later
-			sidecarCount := uint64(len(constructedSidecars))
-			if sidecarCount == 0 {
+			constructedSidecarCount = uint64(len(constructedSidecars))
+			if constructedSidecarCount == 0 {
 				if ctx.Err() != nil {
 					return nil, ctx.Err()
 				}
@@ -212,9 +230,11 @@ func (s *Service) processDataColumnSidecarsFromExecution(ctx context.Context, so
 				continue
 			}
 
+			dataColumnsRecoveredFromELTotal.Inc()
+
 			// Boundary check.
-			if sidecarCount != fieldparams.NumberOfColumns {
-				return nil, errors.Errorf("reconstruct data column sidecars returned %d sidecars, expected %d - should never happen", sidecarCount, fieldparams.NumberOfColumns)
+			if constructedSidecarCount != fieldparams.NumberOfColumns {
+				return nil, errors.Errorf("reconstruct data column sidecars returned %d sidecars, expected %d - should never happen", constructedSidecarCount, fieldparams.NumberOfColumns)
 			}
 
 			unseenIndices, err := s.broadcastAndReceiveUnseenDataColumnSidecars(ctx, source.Slot(), source.ProposerIndex(), columnIndicesToSample, constructedSidecars)
@@ -222,19 +242,12 @@ func (s *Service) processDataColumnSidecarsFromExecution(ctx context.Context, so
 				return nil, errors.Wrap(err, "broadcast and receive unseen data column sidecars")
 			}
 
-			if len(unseenIndices) > 0 {
-				dataColumnsRecoveredFromELTotal.Inc()
+			log.WithFields(logrus.Fields{
+				"count":   len(unseenIndices),
+				"indices": helpers.SortedPrettySliceFromMap(unseenIndices),
+			}).Debug("Constructed data column sidecars from the execution client")
 
-				log.WithFields(logrus.Fields{
-					"root":          fmt.Sprintf("%#x", source.Root()),
-					"slot":          source.Slot(),
-					"proposerIndex": source.ProposerIndex(),
-					"iteration":     iteration,
-					"type":          source.Type(),
-					"count":         len(unseenIndices),
-					"indices":       helpers.SortedPrettySliceFromMap(unseenIndices),
-				}).Debug("Constructed data column sidecars from the execution client")
-			}
+			dataColumnSidecarsObtainedViaELCount.Observe(float64(len(unseenIndices)))
 
 			return nil, nil
 		}

beacon-chain/sync/validate_data_column.go

@@ -51,14 +51,12 @@ func (s *Service) validateDataColumn(ctx context.Context, pid peer.ID, msg *pubs
 	// Decode the message, reject if it fails.
 	m, err := s.decodePubsubMessage(msg)
 	if err != nil {
-		log.WithError(err).Error("Failed to decode message")
 		return pubsub.ValidationReject, err
 	}
 
 	// Reject messages that are not of the expected type.
 	dcsc, ok := m.(*eth.DataColumnSidecar)
 	if !ok {
-		log.WithField("message", m).Error("Message is not of type *eth.DataColumnSidecar")
 		return pubsub.ValidationReject, errWrongMessage
 	}
 

beacon-chain/sync/validate_light_client_test.go

@@ -54,11 +54,13 @@ func TestValidateLightClientOptimisticUpdate(t *testing.T) {
 	cfg.CapellaForkEpoch = 3
 	cfg.DenebForkEpoch = 4
 	cfg.ElectraForkEpoch = 5
+	cfg.FuluForkEpoch = 6
 	cfg.ForkVersionSchedule[[4]byte{1, 0, 0, 0}] = 1
 	cfg.ForkVersionSchedule[[4]byte{2, 0, 0, 0}] = 2
 	cfg.ForkVersionSchedule[[4]byte{3, 0, 0, 0}] = 3
 	cfg.ForkVersionSchedule[[4]byte{4, 0, 0, 0}] = 4
 	cfg.ForkVersionSchedule[[4]byte{5, 0, 0, 0}] = 5
+	cfg.ForkVersionSchedule[[4]byte{6, 0, 0, 0}] = 6
 	params.OverrideBeaconConfig(cfg)
 
 	secondsPerSlot := int(params.BeaconConfig().SecondsPerSlot)
@@ -101,7 +103,10 @@ func TestValidateLightClientOptimisticUpdate(t *testing.T) {
 	}
 
 	for _, test := range tests {
-		for v := 1; v < 6; v++ {
+		for v := range version.All() {
+			if v == version.Phase0 {
+				continue
+			}
 			t.Run(test.name+"_"+version.String(v), func(t *testing.T) {
 				ctx := t.Context()
 				p := p2ptest.NewTestP2P(t)
@@ -180,11 +185,13 @@ func TestValidateLightClientFinalityUpdate(t *testing.T) {
 	cfg.CapellaForkEpoch = 3
 	cfg.DenebForkEpoch = 4
 	cfg.ElectraForkEpoch = 5
+	cfg.FuluForkEpoch = 6
 	cfg.ForkVersionSchedule[[4]byte{1, 0, 0, 0}] = 1
 	cfg.ForkVersionSchedule[[4]byte{2, 0, 0, 0}] = 2
 	cfg.ForkVersionSchedule[[4]byte{3, 0, 0, 0}] = 3
 	cfg.ForkVersionSchedule[[4]byte{4, 0, 0, 0}] = 4
 	cfg.ForkVersionSchedule[[4]byte{5, 0, 0, 0}] = 5
+	cfg.ForkVersionSchedule[[4]byte{6, 0, 0, 0}] = 6
 	params.OverrideBeaconConfig(cfg)
 
 	secondsPerSlot := int(params.BeaconConfig().SecondsPerSlot)
@@ -227,7 +234,10 @@ func TestValidateLightClientFinalityUpdate(t *testing.T) {
 	}
 
 	for _, test := range tests {
-		for v := 1; v < 6; v++ {
+		for v := range version.All() {
+			if v == version.Phase0 {
+				continue
+			}
 			t.Run(test.name+"_"+version.String(v), func(t *testing.T) {
 				ctx := t.Context()
 				p := p2ptest.NewTestP2P(t)

beacon-chain/verification/data_column.go

@@ -361,7 +361,7 @@ func (dv *RODataColumnsVerifier) SidecarParentSeen(parentSeen func([fieldparams.
 		}
 
 		if !dv.fc.HasNode(parentRoot) {
-			return columnErrBuilder(errSidecarParentNotSeen)
+			return columnErrBuilder(errors.Wrapf(errSidecarParentNotSeen, "parent root: %#x", parentRoot))
 		}
 	}
 

changelog/SashaMalysehko_fix-return-after-check.md

@@ -0,0 +1,3 @@
+## Fixed
+
+- Fix missing return after version header check in SubmitAttesterSlashingsV2.

changelog/avoid_kzg_send_after_context_cancel.md

@@ -0,0 +1,3 @@
+### Fixed
+
+- Prevent blocked sends to the KZG batch verifier when the caller context is already canceled, avoiding useless queueing and potential hangs.

changelog/bastin_fix-lcp2p-bug.md

@@ -0,0 +1,3 @@
+### Fixed
+
+- Fix the missing fork version object mapping for Fulu in light client p2p.

changelog/fix_kzg_batch_verifier_timeout_deadlock.md

@@ -0,0 +1,3 @@
+### Fixed
+
+- Fix deadlock in data column gossip KZG batch verification when a caller times out preventing result delivery.

changelog/james-prysm_align-atter-pool-apis.md

@@ -0,0 +1,3 @@
+### Changed
+
+- the /eth/v2/beacon/pool/attestations and /eth/v1/beacon/pool/sync_committees now returns a 503 error if the node is still syncing, the rest api is also working in a similar process to gRPC broadcasting immediately now.

changelog/james-prysm_fix-rest-replay-state.md

@@ -0,0 +1,3 @@
+### Fixed 
+
+- fixed replay state issue in rest api caused by attester and sync committee duties endpoints

changelog/james-prysm_skip-e2e-slot1-check.md

@@ -0,0 +1,3 @@
+### Changed
+
+- e2e sync committee evaluator now skips the first slot after startup, we already skip the fork epoch for checks here, this skip only applies on startup, due to altair always from 0 and validators need to warm up.

changelog/manu-agg.md

@@ -0,0 +1,3 @@
+### Changed
+
+- Pending aggregates: When multiple aggregated attestations only differing by the aggregator index are in the pending queue, only process one of them.

changelog/manu-remove-error-logs.md

@@ -0,0 +1,2 @@
+### Changed
+- `validateDataColumn`: Remove error logs.

changelog/manu_reconstruct-metrics.md

@@ -0,0 +1,7 @@
+### Added
+- prometheus histogram `cells_and_proofs_from_structured_computation_milliseconds` to track computation time for cells and proofs from structured blobs.
+- prometheus histogram `get_blobs_v2_latency_milliseconds` to track RPC latency for `getBlobsV2` calls to the execution layer.
+
+### Changed
+- Run `ComputeCellsAndProofsFromFlat` in parallel to improve performance when computing cells and proofs.
+- Run `ComputeCellsAndProofsFromStructured` in parallel to improve performance when computing cells and proofs.

changelog/potuz_return_indices_updateerr.md

@@ -0,0 +1,3 @@
+### Fixed
+
+- Do not error when committee has been computed correctly but updating the cache failed.

changelog/radek_httperror-analyzer.md

@@ -0,0 +1,3 @@
+### Added
+
+- Static analyzer that ensures each `httputil.HandleError` call is followed by a `return` statement.

changelog/radek_use-statefetch-error.md

@@ -0,0 +1,3 @@
+### Ignored
+
+- Use `WriteStateFetchError` in API handlers whenever possible. 

changelog/satushh-pulltip.md

@@ -0,0 +1,3 @@
+### Fixed
+
+- Removed redundant justified checkpoint update in pullTips. 

testing/endtoend/evaluators/validator.go

@@ -262,6 +262,11 @@ func validatorsSyncParticipation(_ *types.EvaluationContext, conns ...*grpc.Clie
 			// Skip fork slot.
 			continue
 		}
+		// Skip slot 1 at genesis - validators need time to ramp up after chain start.
+		// This is a startup timing issue, not a fork transition issue.
+		if b.Block().Slot() == 1 {
+			continue
+		}
 		expectedParticipation := expectedSyncParticipation
 		switch slots.ToEpoch(b.Block().Slot()) {
 		case params.BeaconConfig().AltairForkEpoch:

tools/analyzers/httperror/BUILD.bazel

@@ -0,0 +1,13 @@
+load("@prysm//tools/go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["analyzer.go"],
+    importpath = "github.com/OffchainLabs/prysm/v7/tools/analyzers/httperror",
+    visibility = ["//visibility:public"],
+    deps = [
+        "@org_golang_x_tools//go/analysis:go_default_library",
+        "@org_golang_x_tools//go/analysis/passes/inspect:go_default_library",
+        "@org_golang_x_tools//go/ast/inspector:go_default_library",
+    ],
+)

tools/analyzers/httperror/analyzer.go

@@ -0,0 +1,163 @@
+package httperror
+
+import (
+	"go/ast"
+
+	"golang.org/x/tools/go/analysis"
+	"golang.org/x/tools/go/analysis/passes/inspect"
+	"golang.org/x/tools/go/ast/inspector"
+)
+
+var Analyzer = &analysis.Analyzer{
+	Name: "httperror",
+	Doc:  "Ensures calls to httputil.HandleError are immediately followed by a return statement.",
+	Requires: []*analysis.Analyzer{
+		inspect.Analyzer,
+	},
+}
+
+func Run(pass *analysis.Pass) (any, error) {
+	ins := pass.ResultOf[inspect.Analyzer].(*inspector.Inspector)
+
+	ins.Preorder([]ast.Node{(*ast.FuncDecl)(nil)}, func(n ast.Node) {
+		fn := n.(*ast.FuncDecl)
+		if fn.Body == nil {
+			return
+		}
+		// top-level: there is no next statement after the function body
+		checkBlock(pass, fn, fn.Body, nil)
+	})
+
+	return nil, nil
+}
+
+func init() {
+	Analyzer.Run = Run
+}
+
+// checkBlock inspects all statements inside block.
+// nextAfter is the statement that comes immediately *after* the statement that contains this block,
+// at the enclosing lexical level (or nil if none).
+// Example: for `if ... { ... } ; return`, when checking the if's body the nextAfter will be the top-level return stmt.
+func checkBlock(pass *analysis.Pass, fn *ast.FuncDecl, block *ast.BlockStmt, nextAfter ast.Stmt) {
+	stmts := block.List
+	for i, stmt := range stmts {
+		// compute the next statement after this statement at the same lexical (ancestor) level:
+		// if there is a sibling, use it; otherwise propagate the nextAfter we were given.
+		var nextForThisStmt ast.Stmt
+		if i+1 < len(stmts) {
+			nextForThisStmt = stmts[i+1]
+		} else {
+			nextForThisStmt = nextAfter
+		}
+
+		// Recurse into nested blocks BEFORE checking this stmt's own expr,
+		// but we must pass nextForThisStmt to nested blocks so nested HandleError
+		// will see the correct "next statement after this statement".
+		switch s := stmt.(type) {
+		case *ast.IfStmt:
+			// pass what's next after the whole if-statement down into its bodies
+			if s.Init != nil {
+				// init is a statement (rare), treat it as contained in s; it should use next being the if's body first stmt,
+				// but for our purposes we don't need to inspect s.Init specially beyond nested calls.
+				// We'll just check it with nextForThisStmt as well.
+				checkBlock(pass, fn, &ast.BlockStmt{List: []ast.Stmt{s.Init}}, nextForThisStmt)
+			}
+			if s.Body != nil {
+				checkBlock(pass, fn, s.Body, nextForThisStmt)
+			}
+			if s.Else != nil {
+				switch els := s.Else.(type) {
+				case *ast.BlockStmt:
+					checkBlock(pass, fn, els, nextForThisStmt)
+				case *ast.IfStmt:
+					// else-if: its body will receive the same nextForThisStmt
+					checkBlock(pass, fn, els.Body, nextForThisStmt)
+				}
+			}
+
+		case *ast.ForStmt:
+			if s.Init != nil {
+				checkBlock(pass, fn, &ast.BlockStmt{List: []ast.Stmt{s.Init}}, nextForThisStmt)
+			}
+			if s.Body != nil {
+				checkBlock(pass, fn, s.Body, nextForThisStmt)
+			}
+			if s.Post != nil {
+				checkBlock(pass, fn, &ast.BlockStmt{List: []ast.Stmt{s.Post}}, nextForThisStmt)
+			}
+
+		case *ast.RangeStmt:
+			if s.Body != nil {
+				checkBlock(pass, fn, s.Body, nextForThisStmt)
+			}
+
+		case *ast.BlockStmt:
+			// nested block (e.g. anonymous block) — propagate nextForThisStmt
+			checkBlock(pass, fn, s, nextForThisStmt)
+		}
+
+		// Now check the current statement itself: is it (or does it contain) a direct call to httputil.HandleError?
+		// We only consider ExprStmt that are direct CallExpr to httputil.HandleError.
+		call := findHandleErrorCall(stmt)
+		if call == nil {
+			continue
+		}
+
+		// Determine the actual "next statement after this call" in lexical function order:
+		// - If there is a sibling in the same block after this stmt, that's next.
+		// - Otherwise, next is nextForThisStmt (propagated from ancestor).
+		var nextStmt ast.Stmt
+		if i+1 < len(stmts) {
+			nextStmt = stmts[i+1]
+		} else {
+			nextStmt = nextAfter
+		}
+
+		// If there is a next statement and it's a return -> OK
+		if nextStmt != nil {
+			if _, ok := nextStmt.(*ast.ReturnStmt); ok {
+				// immediately followed (in lexical order) by a return at some nesting level -> OK
+				continue
+			}
+			// otherwise it's not a return (even if it's an if/for etc) -> violation
+			pass.Reportf(stmt.Pos(), "call to httputil.HandleError must be immediately followed by a return statement")
+			continue
+		}
+
+		// If nextStmt == nil, this call is lexically the last statement in the function.
+		// That is allowed only if the function has no result values.
+		if fn.Type.Results == nil || len(fn.Type.Results.List) == 0 {
+			// void function: allowed
+			continue
+		}
+
+		// Non-void function and it's the last statement → violation
+		pass.Reportf(stmt.Pos(), "call to httputil.HandleError must be followed by return because function has return values")
+	}
+}
+
+// findHandleErrorCall returns the call expression if stmt is a direct call to httputil.HandleError(...),
+// otherwise nil. We only match direct ExprStmt -> CallExpr -> SelectorExpr where selector is httputil.HandleError.
+func findHandleErrorCall(stmt ast.Stmt) *ast.CallExpr {
+	es, ok := stmt.(*ast.ExprStmt)
+	if !ok {
+		return nil
+	}
+	call, ok := es.X.(*ast.CallExpr)
+	if !ok {
+		return nil
+	}
+	sel, ok := call.Fun.(*ast.SelectorExpr)
+	if !ok {
+		return nil
+	}
+	pkgIdent, ok := sel.X.(*ast.Ident)
+	if !ok {
+		return nil
+	}
+	if pkgIdent.Name == "httputil" && sel.Sel.Name == "HandleError" {
+		return call
+	}
+	return nil
+}