saveDataColumnSidecarsExistingFile: Parallelize SSZ encoding of data column sidecars.

If returning before waiting, then the `ctx` used in `processSidecarsFromExecutionFromBlock` is canceled as well, stopping early the blobs retrieval from the EL.

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/BUILD.bazel

@@ -38,6 +38,7 @@ go_library(
         "@com_github_prometheus_client_golang//prometheus/promauto:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",
         "@com_github_spf13_afero//:go_default_library",
+        "@org_golang_x_sync//errgroup:go_default_library",
     ],
 )
 

beacon-chain/db/filesystem/data_column.go

@@ -25,6 +25,7 @@ import (
 	"github.com/OffchainLabs/prysm/v7/time/slots"
 	"github.com/pkg/errors"
 	"github.com/spf13/afero"
+	"golang.org/x/sync/errgroup"
 )
 
 const (
@@ -515,6 +516,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.
@@ -620,7 +626,10 @@ func (dcs *DataColumnStorage) saveDataColumnSidecarsExistingFile(filePath string
 	}
 
 	// Create the SSZ encoded data column sidecars.
-	var sszEncodedDataColumnSidecars []byte
+	var sszEncodedDataColumnSidecarsBytes []byte
+
+	// Initialize the count of the saved SSZ encoded data column sidecar.
+	storedCount := uint8(0)
 
 	for {
 		dataColumnSidecars := pullChan(inputDataColumnSidecars)
@@ -628,7 +637,26 @@ func (dcs *DataColumnStorage) saveDataColumnSidecarsExistingFile(filePath string
 			break
 		}
 
-		for _, dataColumnSidecar := range dataColumnSidecars {
+		var wg errgroup.Group
+		sszEncodedDataColumnSidecars := make([][]byte, len(dataColumnSidecars))
+		for i, dataColumnSidecar := range dataColumnSidecars {
+			wg.Go(func() error {
+				// SSZ encode the data column sidecar.
+				sszEncodedDataColumnSidecar, err := dataColumnSidecar.MarshalSSZ()
+				if err != nil {
+					return errors.Wrap(err, "data column sidecar marshal SSZ")
+				}
+
+				sszEncodedDataColumnSidecars[i] = sszEncodedDataColumnSidecar
+				return nil
+			})
+		}
+
+		if err := wg.Wait(); err != nil {
+			return err
+		}
+
+		for i, dataColumnSidecar := range dataColumnSidecars {
 			// Extract the data columns index.
 			dataColumnIndex := dataColumnSidecar.Index
 
@@ -650,10 +678,7 @@ func (dcs *DataColumnStorage) saveDataColumnSidecarsExistingFile(filePath string
 			}
 
 			// SSZ encode the data column sidecar.
-			sszEncodedDataColumnSidecar, err := dataColumnSidecar.MarshalSSZ()
-			if err != nil {
-				return errors.Wrap(err, "data column sidecar marshal SSZ")
-			}
+			sszEncodedDataColumnSidecar := sszEncodedDataColumnSidecars[i]
 
 			// Compute the size of the SSZ encoded data column sidecar.
 			incomingSszEncodedDataColumnSidecarSize := uint32(len(sszEncodedDataColumnSidecar))
@@ -668,8 +693,11 @@ 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...)
+			sszEncodedDataColumnSidecarsBytes = append(sszEncodedDataColumnSidecarsBytes, sszEncodedDataColumnSidecar...)
 		}
 	}
 
@@ -684,17 +712,20 @@ func (dcs *DataColumnStorage) saveDataColumnSidecarsExistingFile(filePath string
 	}
 
 	// Append the SSZ encoded data column sidecars to the end of the file.
-	count, err = file.WriteAt(sszEncodedDataColumnSidecars, metadata.fileSize)
+	count, err = file.WriteAt(sszEncodedDataColumnSidecarsBytes, metadata.fileSize)
 	if err != nil {
 		return errors.Wrap(err, "write SSZ encoded data column sidecars")
 	}
-	if count != len(sszEncodedDataColumnSidecars) {
+	if count != len(sszEncodedDataColumnSidecarsBytes) {
 		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
 }
@@ -707,7 +738,7 @@ func (dcs *DataColumnStorage) saveDataColumnSidecarsNewFile(filePath string, inp
 
 	var (
 		sszEncodedDataColumnSidecarRefSize int
-		sszEncodedDataColumnSidecars       []byte
+		sszEncodedDataColumnSidecarsBytes  []byte
 	)
 
 	// Initialize the count of the saved SSZ encoded data column sidecar.
@@ -719,7 +750,26 @@ func (dcs *DataColumnStorage) saveDataColumnSidecarsNewFile(filePath string, inp
 			break
 		}
 
-		for _, dataColumnSidecar := range dataColumnSidecars {
+		var wg errgroup.Group
+		sszEncodedDataColumnSidecars := make([][]byte, len(dataColumnSidecars))
+		for i, dataColumnSidecar := range dataColumnSidecars {
+			wg.Go(func() error {
+				// SSZ encode the first data column sidecar.
+				sszEncodedDataColumnSidecar, err := dataColumnSidecar.MarshalSSZ()
+				if err != nil {
+					return errors.Wrap(err, "data column sidecar marshal SSZ")
+				}
+
+				sszEncodedDataColumnSidecars[i] = sszEncodedDataColumnSidecar
+				return nil
+			})
+		}
+
+		if err := wg.Wait(); err != nil {
+			return err
+		}
+
+		for i, dataColumnSidecar := range dataColumnSidecars {
 			// Extract the data column index.
 			dataColumnIndex := dataColumnSidecar.Index
 
@@ -742,10 +792,7 @@ func (dcs *DataColumnStorage) saveDataColumnSidecarsNewFile(filePath string, inp
 			storedCount++
 
 			// SSZ encode the first data column sidecar.
-			sszEncodedDataColumnSidecar, err := dataColumnSidecar.MarshalSSZ()
-			if err != nil {
-				return errors.Wrap(err, "data column sidecar marshal SSZ")
-			}
+			sszEncodedDataColumnSidecar := sszEncodedDataColumnSidecars[i]
 
 			// Check if the size of the SSZ encoded data column sidecar is correct.
 			if sszEncodedDataColumnSidecarRefSize != 0 && len(sszEncodedDataColumnSidecar) != sszEncodedDataColumnSidecarRefSize {
@@ -756,7 +803,7 @@ func (dcs *DataColumnStorage) saveDataColumnSidecarsNewFile(filePath string, inp
 			sszEncodedDataColumnSidecarRefSize = len(sszEncodedDataColumnSidecar)
 
 			// Append the first SSZ encoded data column sidecar to the SSZ encoded data column sidecars.
-			sszEncodedDataColumnSidecars = append(sszEncodedDataColumnSidecars, sszEncodedDataColumnSidecar...)
+			sszEncodedDataColumnSidecarsBytes = append(sszEncodedDataColumnSidecarsBytes, sszEncodedDataColumnSidecar...)
 		}
 	}
 
@@ -793,12 +840,12 @@ func (dcs *DataColumnStorage) saveDataColumnSidecarsNewFile(filePath string, inp
 	rawIndices := indices.raw()
 
 	// Concatenate the version, the data column sidecar size, the data column indices and the SSZ encoded data column sidecar.
-	countToWrite := headerSize + len(sszEncodedDataColumnSidecars)
+	countToWrite := headerSize + len(sszEncodedDataColumnSidecarsBytes)
 	bytes := make([]byte, 0, countToWrite)
 	bytes = append(bytes, byte(version))
 	bytes = append(bytes, encodedSszEncodedDataColumnSidecarSize[:]...)
 	bytes = append(bytes, rawIndices[:]...)
-	bytes = append(bytes, sszEncodedDataColumnSidecars...)
+	bytes = append(bytes, sszEncodedDataColumnSidecarsBytes...)
 
 	countWritten, err := file.Write(bytes)
 	if err != nil {
@@ -808,10 +855,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/BUILD.bazel

@@ -40,6 +40,7 @@ go_library(
         "//beacon-chain/state/state-native:go_default_library",
         "//beacon-chain/state/stategen:go_default_library",
         "//beacon-chain/verification:go_default_library",
+        "//cmd/beacon-chain/flags:go_default_library",
         "//config/fieldparams:go_default_library",
         "//config/params:go_default_library",
         "//consensus-types/blocks:go_default_library",

beacon-chain/execution/engine_client.go

@@ -11,6 +11,7 @@ import (
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/core/peerdas"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/execution/types"
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/verification"
+	"github.com/OffchainLabs/prysm/v7/cmd/beacon-chain/flags"
 	fieldparams "github.com/OffchainLabs/prysm/v7/config/fieldparams"
 	"github.com/OffchainLabs/prysm/v7/config/params"
 	"github.com/OffchainLabs/prysm/v7/consensus-types/blocks"
@@ -532,12 +533,23 @@ 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))
 	}
 
+	if flags.Get().DisableGetBlobsV2 {
+		return []*pb.BlobAndProofV2{}, nil
+	}
+
 	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/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/state/state-design.md

@@ -1,438 +0,0 @@
-# Beacon Chain State Design Document
-
-## Package Structure
-
-### `/beacon-chain/state`
-
-This is the top-level state package that defines interfaces only. This package provides a clean API boundary for state access without implementation details. The package follows the interface segregation principle, breaking down access patterns into granular read-only and write-only interfaces (e.g., `ReadOnlyValidators`, `WriteOnlyValidators`, `ReadOnlyBalances`, `WriteOnlyBalances`).
-
-### `/beacon-chain/state/state-native`
-
-This package contains the actual `BeaconState` struct and all concrete implementations of the state interfaces. It also contains thread-safe getters and setters.
-
-### `/beacon-chain/state/fieldtrie`
-
-A dedicated package for field-level merkle trie functionality. The main component of the package is the `FieldTrie` struct which represents a particular field's merkle trie. The package contains several field trie operations such as recomputing tries, copying and transferring them.
-
-### `/beacon-chain/state/stateutil`
-
-Utility package whose main components are various state merkleization functions. Other things contained in this package include field trie helpers, the implementation of shared references, and validator tracking.
-
-## Beacon State Architecture
-
-### `BeaconState` Structure
-
-The `BeaconState` is a single structure that supports all consensus versions. The value of the `version` field determines the active version of the state. It is used extensively throughout the codebase to determine which code path will be executed.
-
-```go
-type BeaconState struct {
-    version int
-    id      uint64 // Used for tracking states in multi-value slices
-
-    // Common fields (all versions)
-    genesisTime           uint64
-    genesisValidatorsRoot [32]byte
-    slot                  primitives.Slot
-    // ...
-
-    // Phase0+ fields
-    eth1Data                  *ethpb.Eth1Data
-    eth1DataVotes             []*ethpb.Eth1Data
-    eth1DepositIndex          uint64
-    slashings                 []uint64
-    previousEpochAttestations []*ethpb.PendingAttestation
-    currentEpochAttestations  []*ethpb.PendingAttestation
-    // ...
-
-    // Altair+ fields
-    previousEpochParticipation []byte
-    currentEpochParticipation  []byte
-    currentSyncCommittee       *ethpb.SyncCommittee
-    nextSyncCommittee          *ethpb.SyncCommittee
-    // ...
-
-    // Fields for other forks...
-
-    // Internal state management
-    lock                  sync.RWMutex
-    dirtyFields           map[types.FieldIndex]bool
-    dirtyIndices          map[types.FieldIndex][]uint64
-    stateFieldLeaves      map[types.FieldIndex]*fieldtrie.FieldTrie
-    rebuildTrie           map[types.FieldIndex]bool
-    valMapHandler         *stateutil.ValidatorMapHandler
-    merkleLayers          [][][]byte
-    sharedFieldReferences map[types.FieldIndex]*stateutil.Reference
-}
-```
-
-### Multi-Value Slices
-
-Several large array fields of the state are implemented as multi-value slices. This is a specialized data structure that enables efficient sharing and modification of slices between states. A multi-value slice is preferred over a regular slice in scenarios where only a small fraction of the array is updated at a time. In such cases, using a multi-value slice results in fewer memory allocations because many values of the slice will be shared between states, whereas with a regular slice changing even a single item results in copying the full slice. Examples include `MultiValueBlockRoots` and `MultiValueBalances`.
-
-**Example**
-```go
-// Create new multi-value slice
-mvBalances := NewMultiValueBalances([]uint64{32000000000, 32000000000, ...})
-
-// Share across states
-state1.balancesMultiValue = mvBalances
-state2 := state1.Copy()  // state2 shares the same mvBalances
-
-// Modify in state2
-state2.UpdateBalancesAtIndex(0, 31000000000)  // This doesn't create a new multi-value slice.
-
-state1.BalanceAtIndex(0) // this returns 32000000000
-state2.BalanceAtIndex(0) // this returns 31000000000
-```
-
-### Getters/Setters
-
-All beacon state getters and setters follow a consistent pattern. All exported methods are protected from concurrent modification using read locks (for getters) or write locks (for setters).
-
-**Getters**
-
-Values are never returned directly from getters. A copy of the value is returned instead.
-
-```go
-func (b *BeaconState) Fork() *ethpb.Fork {
-    if b.fork == nil {
-        return nil
-    }
-
-    b.lock.RLock()
-    defer b.lock.RUnlock()
-
-    return b.forkVal()
-}
-
-func (b *BeaconState) forkVal() *ethpb.Fork {
-    if b.fork == nil {
-        return nil
-    }
-
-    prevVersion := make([]byte, len(b.fork.PreviousVersion))
-    copy(prevVersion, b.fork.PreviousVersion)
-    currVersion := make([]byte, len(b.fork.CurrentVersion))
-    copy(currVersion, b.fork.CurrentVersion)
-    return &ethpb.Fork{
-        PreviousVersion: prevVersion,
-        CurrentVersion:  currVersion,
-        Epoch:           b.fork.Epoch,
-    }
-}
-```
-
-```go
-func (b *BeaconState) CurrentEpochAttestations() ([]*ethpb.PendingAttestation, error) {
-    if b.version != version.Phase0 {
-        return nil, errNotSupported("CurrentEpochAttestations", b.version)
-    }
-
-    if b.currentEpochAttestations == nil {
-        return nil, nil
-    }
-
-    b.lock.RLock()
-    defer b.lock.RUnlock()
-
-    return b.currentEpochAttestationsVal(), nil
-}
-
-func (b *BeaconState) currentEpochAttestationsVal() []*ethpb.PendingAttestation {
-    if b.currentEpochAttestations == nil {
-        return nil
-    }
-
-    res := make([]*ethpb.PendingAttestation, len(b.currentEpochAttestations))
-    for i := range res {
-        res[i] = b.currentEpochAttestations[i].Copy()
-    }
-    return res
-}
-```
-
-In the case of fields backed by multi-value slices, the appropriate methods of the multi-value slice are invoked.
-
-```go
-func (b *BeaconState) StateRoots() [][]byte {
-    b.lock.RLock()
-    defer b.lock.RUnlock()
-
-    roots := b.stateRootsVal()
-    if roots == nil {
-        return nil
-    }
-    return roots.Slice()
-}
-
-func (b *BeaconState) stateRootsVal() customtypes.StateRoots {
-    if b.stateRootsMultiValue == nil {
-        return nil
-    }
-    return b.stateRootsMultiValue.Value(b)
-}
-
-func (b *BeaconState) StateRootAtIndex(idx uint64) ([]byte, error) {
-    b.lock.RLock()
-    defer b.lock.RUnlock()
-
-    if b.stateRootsMultiValue == nil {
-        return nil, nil
-    }
-    r, err := b.stateRootsMultiValue.At(b, idx)
-    if err != nil {
-        return nil, err
-    }
-    return r[:], nil
-}
-```
-
-**Setters**
-
-Whenever a beacon state field is set, it is marked as dirty. This is needed for hash tree root computation so that the cached merkle branch with the old value of the modified field is recomputed using the new value.
-
-```go
-func (b *BeaconState) SetSlot(val primitives.Slot) error {
-    b.lock.Lock()
-    defer b.lock.Unlock()
-
-    b.slot = val
-    b.markFieldAsDirty(types.Slot)
-    return nil
-}
-```
-
-Several fields of the state are shared between states through a `Reference` mechanism. These references are stored in `b.sharedFieldReferences`. Whenever a state is copied, the reference counter for each of these fields is incremented. When a new value for any of these fields is set, the counter for the existing reference is decremented and a new reference is created for that field.
-
-```go
-type Reference struct {
-    refs uint // Reference counter
-    lock sync.RWMutex
-}
-```
-
-```go
-func (b *BeaconState) SetCurrentParticipationBits(val []byte) error {
-    b.lock.Lock()
-    defer b.lock.Unlock()
-
-    if b.version == version.Phase0 {
-        return errNotSupported("SetCurrentParticipationBits", b.version)
-    }
-
-    b.sharedFieldReferences[types.CurrentEpochParticipationBits].MinusRef()
-    b.sharedFieldReferences[types.CurrentEpochParticipationBits] = stateutil.NewRef(1)
-
-    b.currentEpochParticipation = val
-    b.markFieldAsDirty(types.CurrentEpochParticipationBits)
-    return nil
-}
-```
-
-Updating a single value of an array field requires updating `b.dirtyIndices` to ensure the field trie for that particular field is properly recomputed.
-
-```go
-func (b *BeaconState) UpdateBalancesAtIndex(idx primitives.ValidatorIndex, val uint64) error {
-    if err := b.balancesMultiValue.UpdateAt(b, uint64(idx), val); err != nil {
-        return errors.Wrap(err, "could not update balances")
-    }
-
-    b.lock.Lock()
-    defer b.lock.Unlock()
-
-    b.markFieldAsDirty(types.Balances)
-    b.addDirtyIndices(types.Balances, []uint64{uint64(idx)})
-    return nil
-}
-```
-
-As is the case with getters, for fields backed by multi-value slices the appropriate methods of the multi-value slice are invoked when updating field values. The multi-value slice keeps track of states internally, which means the `Reference` construct is unnecessary.
-
-```go
-func (b *BeaconState) SetStateRoots(val [][]byte) error {
-    b.lock.Lock()
-    defer b.lock.Unlock()
-
-    if b.stateRootsMultiValue != nil {
-        b.stateRootsMultiValue.Detach(b)
-    }
-    b.stateRootsMultiValue = NewMultiValueStateRoots(val)
-
-    b.markFieldAsDirty(types.StateRoots)
-    b.rebuildTrie[types.StateRoots] = true
-    return nil
-}
-
-func (b *BeaconState) UpdateStateRootAtIndex(idx uint64, stateRoot [32]byte) error {
-    if err := b.stateRootsMultiValue.UpdateAt(b, idx, stateRoot); err != nil {
-        return errors.Wrap(err, "could not update state roots")
-    }
-
-    b.lock.Lock()
-    defer b.lock.Unlock()
-
-    b.markFieldAsDirty(types.StateRoots)
-    b.addDirtyIndices(types.StateRoots, []uint64{idx})
-    return nil
-}
-```
-
-### Read-Only Validator
-
-There are two ways in which validators can be accessed through getters. One approach is to use the methods `Validators` and `ValidatorAtIndex`, which return a copy of, respectively, the whole validator set or a particular validator. The other approach is to use the `ReadOnlyValidator` construct via the `ValidatorsReadOnly` and `ValidatorAtIndexReadOnly` methods. The `ReadOnlyValidator` structure is a wrapper around the protobuf validator. The advantage of using the read-only version is that no copy of the underlying validator is made, which helps with performance, especially when accessing a large number of validators (e.g. looping through the whole validator registry). Because the read-only wrapper exposes only getters, each of which returns a copy of the validator’s fields, it prevents accidental mutation of the underlying validator.
-
-```go
-type ReadOnlyValidator struct {
-    validator *ethpb.Validator
-}
-
-// Only getter methods, no setters
-func (v *ReadOnlyValidator) PublicKey() [48]byte
-func (v *ReadOnlyValidator) EffectiveBalance() uint64
-func (v *ReadOnlyValidator) Slashed() bool
-// ... etc
-```
-
-**Usage**
-```go
-// Returns ReadOnlyValidator
-validator, err := state.ValidatorAtIndexReadOnly(idx)
-
-// Can read but not modify
-pubkey := validator.PublicKey()
-balance := validator.EffectiveBalance()
-
-// To modify, must get mutable copy
-validatorCopy := validator.Copy()  // Returns *ethpb.Validator
-validatorCopy.EffectiveBalance = newBalance
-state.UpdateValidatorAtIndex(idx, validatorCopy)
-```
-
-## Field Trie System
-
-For a few large arrays, such as block roots or the validator registry, hashing the state field at every slot would be very expensive. To avoid such re-hashing, the underlying merkle trie of the field is maintained and only the branch(es) corresponding to the changed index(es) are recomputed, instead of the whole trie.
-
-Each state version has a list of active fields defined (e.g. `phase0Fields`, `altairFields`), which serves as the basis for field trie creation.
-
-```go
-type FieldTrie struct {
-    *sync.RWMutex
-    reference     *stateutil.Reference     // The number of states this field trie is shared between
-    fieldLayers   [][]*[32]byte            // Merkle trie layers
-    field         types.FieldIndex         // Which field this field trie represents
-    dataType      types.DataType           // Type of field's array
-    length        uint64                   // Maximum number of elements
-    numOfElems    int                      // Number of elements
-    isTransferred bool                     // Whether trie was transferred
-}
-```
-
-The `DataType` enum indicates the type of the field's array. The possible values are:
-- `BasicArray`: Fixed-size arrays (e.g. `blockRoots`)
-- `CompositeArray`: Variable-size arrays (e.g. `validators`)
-- `CompressedArray`: Variable-size arrays that pack multiple elements per trie leaf (e.g. `balances` with 4 elements per leaf)
-
-### Recomputing a Trie
-
-To avoid recomputing the state root every time any value of the state changes, branches of field tries are not recomputed until the state root is actually needed. When it is necessary to recompute a trie, the `RecomputeTrie` function rebuilds the affected branches in the trie according to the provided changed indices. The changed indices of each field are tracked in the `dirtyIndices` field of the beacon state, which is a `map[types.FieldIndex][]uint64`. The recomputation algorithms for fixed-size and variable-size fields are different.
-
-### Transferring a Trie
-
-When it is expected that an older state won't need its trie for recomputation, its trie layers can be transferred directly to a new trie instead of copying them:
-
-```go
-func (f *FieldTrie) TransferTrie() *FieldTrie {
-    f.isTransferred = true
-    nTrie := &FieldTrie{
-        fieldLayers: f.fieldLayers,  // Direct transfer, no copy
-        field:       f.field,
-        dataType:    f.dataType,
-        reference:   stateutil.NewRef(1),
-        // ...
-    }
-    f.fieldLayers = nil  // Zero out original
-    return nTrie
-}
-```
-
-The downside of this approach is that if it becomes necessary to access the older state's trie later on, the whole trie would have to be recreated since it is empty now. This is especially costly for the validator registry and that is why it is always copied instead.
-
-```go
-if fTrie.FieldReference().Refs() > 1 {
-    var newTrie *fieldtrie.FieldTrie
-    // We choose to only copy the validator
-    // trie as it is pretty expensive to regenerate.
-    if index == types.Validators {
-        newTrie = fTrie.CopyTrie()
-    } else {
-        newTrie = fTrie.TransferTrie()
-    }
-    fTrie.FieldReference().MinusRef()
-    b.stateFieldLeaves[index] = newTrie
-    fTrie = newTrie
-}
-```
-
-## Hash Tree Root Computation
-
-The `BeaconState` structure is not a protobuf object, so there are no SSZ-generated methods for its fields. For each state field, there exists a custom method that returns its root. One advantage of doing it this way is the possibility to have the most efficient implementation possible. As an example, when computing the root of the validator registry, validators are hashed in parallel and vectorized sha256 computation is utilized to speed up the computation.
-
-```go
-func OptimizedValidatorRoots(validators []*ethpb.Validator) ([][32]byte, error) {
-    // Exit early if no validators are provided.
-    if len(validators) == 0 {
-        return [][32]byte{}, nil
-    }
-    wg := sync.WaitGroup{}
-    n := runtime.GOMAXPROCS(0)
-    rootsSize := len(validators) * validatorFieldRoots
-    groupSize := len(validators) / n
-    roots := make([][32]byte, rootsSize)
-    wg.Add(n - 1)
-    for j := 0; j < n-1; j++ {
-        go hashValidatorHelper(validators, roots, j, groupSize, &wg)
-    }
-    for i := (n - 1) * groupSize; i < len(validators); i++ {
-        fRoots, err := ValidatorFieldRoots(validators[i])
-        if err != nil {
-            return [][32]byte{}, errors.Wrap(err, "could not compute validators merkleization")
-        }
-        for k, root := range fRoots {
-            roots[i*validatorFieldRoots+k] = root
-        }
-    }
-    wg.Wait()
-
-    // A validator's tree can represented with a depth of 3. As log2(8) = 3
-    // Using this property we can lay out all the individual fields of a
-    // validator and hash them in single level using our vectorized routine.
-    for range validatorTreeDepth {
-        // Overwrite input lists as we are hashing by level
-        // and only need the highest level to proceed.
-        roots = htr.VectorizedSha256(roots)
-    }
-    return roots, nil
-}
-```
-
-Calculating the full state root is very expensive; therefore, it is done in a lazy fashion. Previously generated merkle layers are cached in the state and only merkle trie branches corresponding to dirty indices are regenerated before returning the final root.
-
-```go
-func (b *BeaconState) HashTreeRoot(ctx context.Context) ([32]byte, error) {
-    ctx, span := trace.StartSpan(ctx, "beaconState.HashTreeRoot")
-    defer span.End()
-
-    b.lock.Lock()
-    defer b.lock.Unlock()
-    
-    // When len(b.merkleLayers) > 0, this function returns immediately
-    if err := b.initializeMerkleLayers(ctx); err != nil {
-        return [32]byte{}, err
-    }
-    // Dirty fields are tracked in the beacon state through b.dirtyFields
-    if err := b.recomputeDirtyFields(ctx); err != nil {
-        return [32]byte{}, err
-    }
-    return bytesutil.ToBytes32(b.merkleLayers[len(b.merkleLayers)-1][0]), 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

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"os"
 	"path"
+	"sync"
 	"time"
 
 	"github.com/OffchainLabs/prysm/v7/beacon-chain/blockchain"
@@ -48,7 +49,15 @@ func (s *Service) beaconBlockSubscriber(ctx context.Context, msg proto.Message)
 		return errors.Wrap(err, "new ro block with root")
 	}
 
-	go s.processSidecarsFromExecutionFromBlock(ctx, roBlock)
+	var wg sync.WaitGroup
+	wg.Go(func() {
+		if err := s.processSidecarsFromExecutionFromBlock(ctx, roBlock); err != nil {
+			log.WithError(err).WithFields(logrus.Fields{
+				"root": fmt.Sprintf("%#x", root),
+				"slot": block.Slot(),
+			}).Error("Failed to process sidecars from execution from block")
+		}
+	})
 
 	if err := s.cfg.chain.ReceiveBlock(ctx, signed, root, nil); err != nil {
 		if blockchain.IsInvalidBlock(err) {
@@ -69,28 +78,33 @@ func (s *Service) beaconBlockSubscriber(ctx context.Context, msg proto.Message)
 		}
 		return err
 	}
+
 	if err := s.processPendingAttsForBlock(ctx, root); err != nil {
 		return errors.Wrap(err, "process pending atts for block")
 	}
+
+	wg.Wait()
+
 	return nil
 }
 
 // processSidecarsFromExecutionFromBlock retrieves (if available) sidecars data from the execution client,
 // builds corresponding sidecars, save them to the storage, and broadcasts them over P2P if necessary.
-func (s *Service) processSidecarsFromExecutionFromBlock(ctx context.Context, roBlock blocks.ROBlock) {
+func (s *Service) processSidecarsFromExecutionFromBlock(ctx context.Context, roBlock blocks.ROBlock) error {
 	if roBlock.Version() >= version.Fulu {
 		if err := s.processDataColumnSidecarsFromExecution(ctx, peerdas.PopulateFromBlock(roBlock)); err != nil {
-			log.WithError(err).Error("Failed to process data column sidecars from execution")
-			return
+			return errors.Wrap(err, "process data column sidecars from execution")
 		}
 
-		return
+		return nil
 	}
 
 	if roBlock.Version() >= version.Deneb {
 		s.processBlobSidecarsFromExecution(ctx, roBlock)
-		return
+		return nil
 	}
+
+	return nil
 }
 
 // processBlobSidecarsFromExecution retrieves (if available) blob sidecars data from the execution client,
@@ -168,7 +182,6 @@ func (s *Service) processDataColumnSidecarsFromExecution(ctx context.Context, so
 	key := fmt.Sprintf("%#x", source.Root())
 	if _, err, _ := s.columnSidecarsExecSingleFlight.Do(key, func() (any, error) {
 		const delay = 250 * time.Millisecond
-		secondsPerHalfSlot := time.Duration(params.BeaconConfig().SecondsPerSlot/2) * time.Second
 
 		commitments, err := source.Commitments()
 		if err != nil {
@@ -186,15 +199,35 @@ func (s *Service) processDataColumnSidecarsFromExecution(ctx context.Context, so
 			return nil, errors.Wrap(err, "column indices to sample")
 		}
 
-		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
 			}
 
+			// Return if the context is done.
+			if ctx.Err() != nil {
+				return nil, ctx.Err()
+			}
+
+			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,19 +235,11 @@ func (s *Service) processDataColumnSidecarsFromExecution(ctx context.Context, so
 			}
 
 			// No sidecars are retrieved from the EL, retry later
-			sidecarCount := uint64(len(constructedSidecars))
-			if sidecarCount == 0 {
-				if ctx.Err() != nil {
-					return nil, ctx.Err()
-				}
-
-				time.Sleep(delay)
-				continue
-			}
+			constructedCount := uint64(len(constructedSidecars))
 
 			// 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 > 0 && 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,7 +247,7 @@ func (s *Service) processDataColumnSidecarsFromExecution(ctx context.Context, so
 				return nil, errors.Wrap(err, "broadcast and receive unseen data column sidecars")
 			}
 
-			if len(unseenIndices) > 0 {
+			if constructedCount > 0 {
 				dataColumnsRecoveredFromELTotal.Inc()
 
 				log.WithFields(logrus.Fields{
@@ -234,9 +259,12 @@ func (s *Service) processDataColumnSidecarsFromExecution(ctx context.Context, so
 					"count":         len(unseenIndices),
 					"indices":       helpers.SortedPrettySliceFromMap(unseenIndices),
 				}).Debug("Constructed data column sidecars from the execution client")
+
+				return nil, nil
 			}
 
-			return nil, nil
+			// Wait before retrying.
+			time.Sleep(delay)
 		}
 	}); err != nil {
 		return err
@@ -271,6 +299,11 @@ func (s *Service) broadcastAndReceiveUnseenDataColumnSidecars(
 		unseenIndices[sidecar.Index] = true
 	}
 
+	// Exit early if there are no nothing to broadcast or receive.
+	if len(unseenSidecars) == 0 {
+		return nil, nil
+	}
+
 	// Broadcast all the data column sidecars we reconstructed but did not see via gossip (non blocking).
 	if err := s.cfg.p2p.BroadcastDataColumnSidecars(ctx, unseenSidecars); err != nil {
 		return nil, errors.Wrap(err, "broadcast data column sidecars")

beacon-chain/sync/subscriber_beacon_blocks_test.go

@@ -194,7 +194,8 @@ func TestProcessSidecarsFromExecutionFromBlock(t *testing.T) {
 					},
 					seenBlobCache: lruwrpr.New(1),
 				}
-				s.processSidecarsFromExecutionFromBlock(t.Context(), roBlock)
+				err := s.processSidecarsFromExecutionFromBlock(t.Context(), roBlock)
+				require.NoError(t, err)
 				require.Equal(t, tt.expectedBlobCount, len(chainService.Blobs))
 			})
 		}
@@ -293,7 +294,8 @@ func TestProcessSidecarsFromExecutionFromBlock(t *testing.T) {
 				roBlock, err := blocks.NewROBlock(sb)
 				require.NoError(t, err)
 
-				s.processSidecarsFromExecutionFromBlock(t.Context(), roBlock)
+				err = s.processSidecarsFromExecutionFromBlock(t.Context(), roBlock)
+				require.NoError(t, err)
 				require.Equal(t, tt.expectedDataColumnCount, len(chainService.DataColumns))
 			})
 		}

beacon-chain/sync/subscriber_data_column_sidecar.go

@@ -25,12 +25,12 @@ func (s *Service) dataColumnSubscriber(ctx context.Context, msg proto.Message) e
 	}
 
 	if err := s.receiveDataColumnSidecar(ctx, sidecar); err != nil {
-		return errors.Wrap(err, "receive data column sidecar")
+		return wrapDataColumnError(sidecar, "receive data column sidecar", err)
 	}
 
 	wg.Go(func() error {
 		if err := s.processDataColumnSidecarsFromReconstruction(ctx, sidecar); err != nil {
-			return errors.Wrap(err, "process data column sidecars from reconstruction")
+			return wrapDataColumnError(sidecar, "process data column sidecars from reconstruction", err)
 		}
 
 		return nil
@@ -38,7 +38,7 @@ func (s *Service) dataColumnSubscriber(ctx context.Context, msg proto.Message) e
 
 	wg.Go(func() error {
 		if err := s.processDataColumnSidecarsFromExecution(ctx, peerdas.PopulateFromSidecar(sidecar)); err != nil {
-			return errors.Wrap(err, "process data column sidecars from execution")
+			return wrapDataColumnError(sidecar, "process data column sidecars from execution", err)
 		}
 
 		return nil
@@ -110,3 +110,7 @@ func (s *Service) allDataColumnSubnets(_ primitives.Slot) map[uint64]bool {
 
 	return allSubnets
 }
+
+func wrapDataColumnError(sidecar blocks.VerifiedRODataColumn, message string, err error) error {
+	return fmt.Errorf("%s - slot %d, root %s: %w", message, sidecar.SignedBlockHeader.Header.Slot, fmt.Sprintf("%#x", sidecar.BlockRoot()), err)
+}

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_disable_get_blobs_v2.md

@@ -0,0 +1,2 @@
+### Added
+- `--disable-get-blobs-v2` flag.

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_state-design-doc.md

@@ -1,3 +0,0 @@
-### Added
-
-- Add state design doc.

changelog/radek_use-statefetch-error.md

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

cmd/beacon-chain/flags/base.go

@@ -356,4 +356,9 @@ var (
 		Usage: "A comma-separated list of exponents (of 2) in decreasing order, defining the state diff hierarchy levels. The last exponent must be greater than or equal to 5.",
 		Value: cli.NewIntSlice(21, 18, 16, 13, 11, 9, 5),
 	}
+	// DisableGetBlobsV2 disables the engine_getBlobsV2 usage.
+	DisableGetBlobsV2 = &cli.BoolFlag{
+		Name:  "disable-get-blobs-v2",
+		Usage: "Disables the engine_getBlobsV2 usage.",
+	}
 )

cmd/beacon-chain/flags/config.go

@@ -17,6 +17,7 @@ type GlobalFlags struct {
 	SubscribeToAllSubnets           bool
 	Supernode                       bool
 	SemiSupernode                   bool
+	DisableGetBlobsV2               bool
 	MinimumSyncPeers                int
 	MinimumPeersPerSubnet           int
 	MaxConcurrentDials              int
@@ -72,6 +73,11 @@ func ConfigureGlobalFlags(ctx *cli.Context) error {
 		cfg.SemiSupernode = true
 	}
 
+	if ctx.Bool(DisableGetBlobsV2.Name) {
+		log.Warning("Disabling `engine_getBlobsV2` API")
+		cfg.DisableGetBlobsV2 = true
+	}
+
 	// State-diff-exponents
 	cfg.StateDiffExponents = ctx.IntSlice(StateDiffExponents.Name)
 	if features.Get().EnableStateDiff {

cmd/beacon-chain/main.go

@@ -147,6 +147,7 @@ var appFlags = []cli.Flag{
 	flags.SlasherDirFlag,
 	flags.SlasherFlag,
 	flags.JwtId,
+	flags.DisableGetBlobsV2,
 	storage.BlobStoragePathFlag,
 	storage.DataColumnStoragePathFlag,
 	storage.BlobStorageLayout,

cmd/beacon-chain/usage.go

@@ -169,6 +169,7 @@ var appHelpFlagGroups = []flagGroup{
 			flags.ExecutionJWTSecretFlag,
 			flags.JwtId,
 			flags.InteropMockEth1DataVotesFlag,
+			flags.DisableGetBlobsV2,
 		},
 	},
 	{ // Flags relevant to configuring beacon chain monitoring.

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
+}