Add Pyroscope continuous profiling

This PR hardens the pending payload structure on a few ways:

- It quickly ignores before signature verification if we already have
many pending payloads
- It ignores rather than reject payloads for the self-build block if
they fail verification, this is because we may get as in devnet-zero
from gossip payloads from unknown branches.
- It adds a counter to prevent being DOS by the previous feature. Since
we are ignoring instead of rejecting those payloads, it only allows 3
such payloads to come. This counter is pruned on finalization.

---------

Co-authored-by: terence <terence@prysmaticlabs.com>

beacon-chain/sync/pending_payload_envelope.go

@@ -11,6 +11,7 @@ import (
 const (
 	maxPendingPayloadRoots    = 128
 	maxPendingBuildersPerRoot = 2
+	maxSelfBuildSigFailures   = 3
 )
 
 // processPendingPayloadEnvelopeQueue sweeps the pending envelope map at
@@ -87,12 +88,17 @@ func (s *Service) prunePendingPayloadEnvelopes() {
 	defer s.pendingEnvelopeLock.Unlock()
 
 	finalizedEpoch := s.cfg.chain.FinalizedCheckpt().Epoch
+	deleted := false
 	for root, inner := range s.pendingPayloadEnvelopes {
 		for _, env := range inner {
 			if slots.ToEpoch(env.Message.Slot) < finalizedEpoch {
 				delete(s.pendingPayloadEnvelopes, root)
+				deleted = true
 			}
 			break // only need one envelope per root; admission enforces current-slot
 		}
 	}
+	if deleted {
+		s.selfBuildSigFailures = 0
+	}
 }

beacon-chain/sync/pending_payload_envelope_test.go

@@ -298,11 +298,19 @@ func TestQueuePendingPayloadEnvelope_SelfBuildInLookaheadVerifiesSignature(t *te
 	env, err := e.Envelope()
 	require.NoError(t, err)
 
-	// Self-build in the same epoch (lookahead) still verifies the signature.
+	// Self-build in the same epoch (lookahead) verifies the signature but ignores failures.
 	v := &mockExecutionPayloadEnvelopeVerifier{errSignature: errors.New("bad signature")}
 	result, err := s.queuePendingPayloadEnvelope(ctx, v, env, signedEnv)
-	require.NotNil(t, err)
-	require.Equal(t, pubsub.ValidationReject, result)
+	require.NoError(t, err)
+	require.Equal(t, pubsub.ValidationIgnore, result)
+	require.Equal(t, 1, s.selfBuildSigFailures)
+
+	// After maxSelfBuildSigFailures, skip the signature check entirely and queue the envelope.
+	s.selfBuildSigFailures = maxSelfBuildSigFailures
+	result, err = s.queuePendingPayloadEnvelope(ctx, v, env, signedEnv)
+	require.NoError(t, err)
+	require.Equal(t, pubsub.ValidationIgnore, result)
+	require.Equal(t, maxSelfBuildSigFailures, s.selfBuildSigFailures)
 }
 
 func TestQueuePendingPayloadEnvelope_RejectBadSignature(t *testing.T) {

beacon-chain/sync/service.go

@@ -197,6 +197,7 @@ type Service struct {
 	newExecutionPayloadEnvelopeVerifier verification.NewExecutionPayloadEnvelopeVerifier
 	pendingPayloadEnvelopes             map[[32]byte]map[uint64]*ethpb.SignedExecutionPayloadEnvelope
 	pendingEnvelopeLock                 sync.RWMutex
+	selfBuildSigFailures                int
 }
 
 // NewService initializes new regular sync service.

beacon-chain/sync/validate_execution_payload_envelope.go

@@ -152,43 +152,55 @@ func (s *Service) queuePendingPayloadEnvelope(
 	proposerInLookahead := (stateEpoch == currentEpoch || stateEpoch+1 == currentEpoch)
 	builderIdx := uint64(env.BuilderIndex())
 	isSelfBuild := builderIdx == uint64(params.BeaconConfig().BuilderIndexSelfBuild)
+	root := env.BeaconBlockRoot()
+	s.pendingEnvelopeLock.Lock()
+	defer s.pendingEnvelopeLock.Unlock()
+	inner, rootExists := s.pendingPayloadEnvelopes[root]
+	if !isSelfBuild && len(s.pendingPayloadEnvelopes) >= maxPendingPayloadRoots {
+		log.Debug("Too many pending payload roots, ignoring new payload envelope")
+		return pubsub.ValidationIgnore, nil
+	}
+	if !isSelfBuild && len(inner) >= maxPendingBuildersPerRoot {
+		log.Debug("Too many pending builders for root, ignoring new payload envelope")
+		return pubsub.ValidationIgnore, nil
+	}
+
+	if isSelfBuild && s.selfBuildSigFailures >= maxSelfBuildSigFailures {
+		log.Debug("Ignoring self-built payload envelope because of too many signature failures")
+		return pubsub.ValidationIgnore, nil
+	}
+
 	if !isSelfBuild || proposerInLookahead {
 		if err := v.VerifySignature(st); err != nil {
-			return pubsub.ValidationReject, err
+			if isSelfBuild {
+				s.selfBuildSigFailures++
+				log.WithError(err).Debug("Ignoring self-built payload with invalid signature")
+				return pubsub.ValidationIgnore, nil
+			} else {
+				return pubsub.ValidationReject, err
+			}
 		}
 	} else {
 		log.Debug("Ignoring payload envelope from self-build outside of the Lookahead window")
 		return pubsub.ValidationIgnore, nil
 	}
-	root := env.BeaconBlockRoot()
-	s.pendingEnvelopeLock.Lock()
-	inner, rootExists := s.pendingPayloadEnvelopes[root]
 	if !rootExists {
-		if !isSelfBuild && len(s.pendingPayloadEnvelopes) >= maxPendingPayloadRoots {
-			s.pendingEnvelopeLock.Unlock()
-			return pubsub.ValidationIgnore, nil
-		}
 		inner = make(map[uint64]*ethpb.SignedExecutionPayloadEnvelope)
 		s.pendingPayloadEnvelopes[root] = inner
 	} else {
 		for _, existing := range inner {
 			if existing.Message.Slot != signedEnvelope.Message.Slot {
-				s.pendingEnvelopeLock.Unlock()
+				log.Debug("Ignoring payload envelope with mismatched slot")
 				return pubsub.ValidationIgnore, nil
 			}
 			break
 		}
 	}
 	if _, exists := inner[builderIdx]; exists {
-		s.pendingEnvelopeLock.Unlock()
-		return pubsub.ValidationIgnore, nil
-	}
-	if !isSelfBuild && len(inner) >= maxPendingBuildersPerRoot {
-		s.pendingEnvelopeLock.Unlock()
+		log.Debug("Already have a pending payload envelope for this builder and root, ignoring")
 		return pubsub.ValidationIgnore, nil
 	}
 	inner[builderIdx] = signedEnvelope
-	s.pendingEnvelopeLock.Unlock()
 
 	s.pendingQueueLock.RLock()
 	inPendingQueue := s.seenPendingBlocks[root]

beacon-chain/sync/validate_execution_payload_envelope_test.go

@@ -266,6 +266,68 @@ func envelopeToPubsub(t *testing.T, s *Service, p p2p.P2P, env *ethpb.SignedExec
 	}
 }
 
+func TestQueuePendingPayloadEnvelope_SelfBuildInvalidSignature(t *testing.T) {
+	ctx := context.Background()
+	tests := []struct {
+		name       string
+		builderIdx primitives.BuilderIndex
+		result     pubsub.ValidationResult
+		wantError  bool
+	}{
+		{
+			name:       "self-build with invalid signature is ignored",
+			builderIdx: params.BeaconConfig().BuilderIndexSelfBuild,
+			result:     pubsub.ValidationIgnore,
+		},
+		{
+			name:       "non-self-build with invalid signature is rejected",
+			builderIdx: 42,
+			result:     pubsub.ValidationReject,
+			wantError:  true,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			p := p2ptest.NewTestP2P(t)
+			chainService := &mock.ChainService{
+				Genesis:             time.Unix(time.Now().Unix()-int64(params.BeaconConfig().SecondsPerSlot), 0),
+				FinalizedCheckPoint: &ethpb.Checkpoint{},
+			}
+			st, err := util.NewBeaconStateFulu()
+			require.NoError(t, err)
+			chainService.State = st
+
+			s := &Service{
+				seenPayloadEnvelopeCache: lruwrpr.New(10),
+				pendingPayloadEnvelopes:  make(map[[32]byte]map[uint64]*ethpb.SignedExecutionPayloadEnvelope),
+				cfg: &config{
+					p2p:         p,
+					initialSync: &mockSync.Sync{},
+					chain:       chainService,
+					clock:       startup.NewClock(chainService.Genesis, chainService.ValidatorsRoot),
+				},
+			}
+			s.newExecutionPayloadEnvelopeVerifier = testNewExecutionPayloadEnvelopeVerifier(mockExecutionPayloadEnvelopeVerifier{
+				errBlockRootSeen: errors.New("not seen"),
+				errSignature:     errors.New("bad signature"),
+			})
+
+			root := [32]byte{0x01}
+			blockHash := [32]byte{0x02}
+			env := testSignedExecutionPayloadEnvelope(t, 1, tc.builderIdx, root, blockHash)
+			msg := envelopeToPubsub(t, s, p, env)
+
+			result, err := s.validateExecutionPayloadEnvelope(ctx, "", msg)
+			if tc.wantError {
+				require.NotNil(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+			require.Equal(t, tc.result, result)
+		})
+	}
+}
+
 func testSignedExecutionPayloadEnvelope(t *testing.T, slot primitives.Slot, builderIdx primitives.BuilderIndex, root, blockHash [32]byte) *ethpb.SignedExecutionPayloadEnvelope {
 	t.Helper()
 

changelog/manu_pyroscope.md

@@ -0,0 +1,3 @@
+### Added
+
+- Pyroscope continuous profiling support for the beacon node via `--pyroscope`, `--pyroscope-server`, and `--pyroscope-app-name` flags.

changelog/potuz_self_built_signature.md

@@ -0,0 +1,2 @@
+### Added
+- Ignore self-built payloads with invalid signatures.

cmd/beacon-chain/main.go

@@ -132,6 +132,9 @@ var appFlags = []cli.Flag{
 	debug.MemProfileRateFlag,
 	debug.BlockProfileRateFlag,
 	debug.MutexProfileFractionFlag,
+	debug.PyroscopeFlag,
+	debug.PyroscopeServerFlag,
+	debug.PyroscopeAppNameFlag,
 	cmd.LogFileName,
 	cmd.EnableUPnPFlag,
 	cmd.ConfigFileFlag,

cmd/beacon-chain/usage.go

@@ -231,6 +231,9 @@ var appHelpFlagGroups = []flagGroup{
 			debug.PProfAddrFlag,
 			debug.PProfFlag,
 			debug.PProfPortFlag,
+			debug.PyroscopeFlag,
+			debug.PyroscopeServerFlag,
+			debug.PyroscopeAppNameFlag,
 			flags.SetGCPercent,
 		},
 	},

deps.bzl

@@ -1349,6 +1349,18 @@ def prysm_deps():
         sum = "h1:d2/eIbH9XjD1fFwD5SHv8x168fjbQ9PB8hvs8DSEC08=",
         version = "v0.3.1-0.20210208050101-bfb5c8eec0e4",
     )
+    go_repository(
+        name = "com_github_grafana_pyroscope_go",
+        importpath = "github.com/grafana/pyroscope-go",
+        sum = "h1:VWBBlqxjyR0Cwk2W6UrE8CdcdD80GOFNutj0Kb1T8ac=",
+        version = "v1.2.7",
+    )
+    go_repository(
+        name = "com_github_grafana_pyroscope_go_godeltaprof",
+        importpath = "github.com/grafana/pyroscope-go/godeltaprof",
+        sum = "h1:c1Us8i6eSmkW+Ez05d3co8kasnuOY813tbMN8i/a3Og=",
+        version = "v0.1.9",
+    )
     go_repository(
         name = "com_github_graph_gophers_graphql_go",
         importpath = "github.com/graph-gophers/graphql-go",

go.mod

@@ -28,6 +28,7 @@ require (
 	github.com/google/gofuzz v1.2.0
 	github.com/google/uuid v1.6.0
 	github.com/gostaticanalysis/comment v1.4.2
+	github.com/grafana/pyroscope-go v1.2.7
 	github.com/grpc-ecosystem/go-grpc-middleware v1.2.2
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
 	github.com/hashicorp/golang-lru v0.5.5-0.20210104140557-80c98217689d
@@ -154,6 +155,7 @@ require (
 	github.com/google/gopacket v1.1.19 // indirect
 	github.com/google/pprof v0.0.0-20250202011525-fc3143867406 // indirect
 	github.com/gorilla/websocket v1.5.3 // indirect
+	github.com/grafana/pyroscope-go/godeltaprof v0.1.9 // indirect
 	github.com/graph-gophers/graphql-go v1.3.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.25.1 // indirect
 	github.com/hashicorp/go-bexpr v0.1.10 // indirect

go.sum

@@ -440,6 +440,10 @@ github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/ad
 github.com/gostaticanalysis/comment v1.4.2 h1:hlnx5+S2fY9Zo9ePo4AhgYsYHbM2+eAv8m/s1JiCd6Q=
 github.com/gostaticanalysis/comment v1.4.2/go.mod h1:KLUTGDv6HOCotCH8h2erHKmpci2ZoR8VPu34YA2uzdM=
 github.com/gostaticanalysis/testutil v0.3.1-0.20210208050101-bfb5c8eec0e4/go.mod h1:D+FIZ+7OahH3ePw/izIEeH5I06eKs1IKI4Xr64/Am3M=
+github.com/grafana/pyroscope-go v1.2.7 h1:VWBBlqxjyR0Cwk2W6UrE8CdcdD80GOFNutj0Kb1T8ac=
+github.com/grafana/pyroscope-go v1.2.7/go.mod h1:o/bpSLiJYYP6HQtvcoVKiE9s5RiNgjYTj1DhiddP2Pc=
+github.com/grafana/pyroscope-go/godeltaprof v0.1.9 h1:c1Us8i6eSmkW+Ez05d3co8kasnuOY813tbMN8i/a3Og=
+github.com/grafana/pyroscope-go/godeltaprof v0.1.9/go.mod h1:2+l7K7twW49Ct4wFluZD3tZ6e0SjanjcUUBPVD/UuGU=
 github.com/graph-gophers/graphql-go v1.3.0 h1:Eb9x/q6MFpCLz7jBCiP/WTxjSDrYLR1QY41SORZyNJ0=
 github.com/graph-gophers/graphql-go v1.3.0/go.mod h1:9CQHMSxwO4MprSdzoIEobiHpoLtHm77vfxsvsIN5Vuc=
 github.com/gregjones/httpcache v0.0.0-20170920190843-316c5e0ff04e/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=

runtime/debug/BUILD.bazel

@@ -18,6 +18,7 @@ go_library(
     importpath = "github.com/OffchainLabs/prysm/v7/runtime/debug",
     visibility = ["//visibility:public"],
     deps = [
+        "@com_github_grafana_pyroscope_go//:go_default_library",
         "@com_github_prometheus_client_golang//prometheus:go_default_library",
         "@com_github_prometheus_client_golang//prometheus/promauto:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",

runtime/debug/debug.go

@@ -37,6 +37,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/grafana/pyroscope-go"
 	log "github.com/sirupsen/logrus"
 	"github.com/urfave/cli/v2"
 )
@@ -78,6 +79,23 @@ var (
 		Name:  "blockprofilerate",
 		Usage: "Turns on block profiling with the given rate.",
 	}
+	// PyroscopeFlag to enable Pyroscope continuous profiling.
+	PyroscopeFlag = &cli.BoolFlag{
+		Name:  "pyroscope",
+		Usage: "Enables Pyroscope continuous profiling.",
+	}
+	// PyroscopeServerFlag to specify the Pyroscope server URL.
+	PyroscopeServerFlag = &cli.StringFlag{
+		Name:  "pyroscope-server",
+		Usage: "Pyroscope server URL for continuous profiling.",
+		Value: "http://localhost:4040",
+	}
+	// PyroscopeAppNameFlag to specify the application name tag.
+	PyroscopeAppNameFlag = &cli.StringFlag{
+		Name:  "pyroscope-app-name",
+		Usage: "Application name tag for Pyroscope profiling.",
+		Value: "prysm.beacon-chain",
+	}
 )
 
 // HandlerT implements the debugging API.
@@ -323,9 +341,42 @@ func Setup(ctx *cli.Context) error {
 		address := fmt.Sprintf("%s:%d", ctx.String(PProfAddrFlag.Name), ctx.Int(PProfPortFlag.Name))
 		startPProf(address)
 	}
+	// Pyroscope continuous profiling
+	if ctx.Bool(PyroscopeFlag.Name) {
+		startPyroscope(ctx)
+	}
 	return nil
 }
 
+func startPyroscope(ctx *cli.Context) {
+	serverAddress := ctx.String(PyroscopeServerFlag.Name)
+	appName := ctx.String(PyroscopeAppNameFlag.Name)
+	_, err := pyroscope.Start(pyroscope.Config{
+		ApplicationName: appName,
+		ServerAddress:   serverAddress,
+		DisableGCRuns:   true,
+		ProfileTypes: []pyroscope.ProfileType{
+			pyroscope.ProfileCPU,
+			pyroscope.ProfileAllocObjects,
+			pyroscope.ProfileAllocSpace,
+			pyroscope.ProfileInuseObjects,
+			pyroscope.ProfileInuseSpace,
+			pyroscope.ProfileGoroutines,
+			pyroscope.ProfileMutexCount,
+			pyroscope.ProfileMutexDuration,
+			pyroscope.ProfileBlockCount,
+			pyroscope.ProfileBlockDuration,
+		},
+	})
+
+	if err != nil {
+		log.WithError(err).Error("Failed to start Pyroscope profiling")
+		return
+	}
+
+	log.WithField("server", serverAddress).WithField("app", appName).Info("Started Pyroscope continuous profiling")
+}
+
 func startPProf(address string) {
 	log.WithField("addr", fmt.Sprintf("http://%s/debug/pprof", address)).Info("Starting pprof server")
 	go func() {