From 3da2e223fa5650cd7710f1964b53926788d4da7a Mon Sep 17 00:00:00 2001 From: Oleksandr Lystopad Date: Mon, 17 Aug 2026 14:09:26 +0200 Subject: [PATCH 1/5] cl/beacon: report a failing payload poll once, and an unregistered fee recipient at all Polling retries at the retry cadence, so one slot whose payload never arrived reported itself once per attempt and buried everything else in that slot. It is reported when the window is over, once, with the slot, how many attempts failed and the first reason, which is the one that says what went wrong. A window that recovers reports nothing, and neither does a caller that went away: its own cancellation comes back through the collection call, and that is the slot ending rather than the execution layer failing. A failure that happened before the caller left is still reported, since giving up is often the consequence of it. Production falls back to the zero address when nothing is registered for the proposer, which gives that block's fees away, and said nothing. It now says so once per proposer, claimed in one step so that requests for the same slot arriving together do not each decide they are the first, and bounded so an unregistered proposer cannot accumulate. --- cl/beacon/handler/block_production.go | 52 +++++- cl/beacon/handler/block_production_test.go | 182 ++++++++++++++++++++- cl/beacon/handler/handler.go | 10 +- 3 files changed, 234 insertions(+), 10 deletions(-) diff --git a/cl/beacon/handler/block_production.go b/cl/beacon/handler/block_production.go index 3d156ef8dbc..04f9c6d4d10 100644 --- a/cl/beacon/handler/block_production.go +++ b/cl/beacon/handler/block_production.go @@ -289,6 +289,27 @@ func (a *ApiHandler) expectedWithdrawals( return cltypes.ConvertConsensusWithdrawalsToExecutionWithdrawals(consensusWithdrawals), nil } +// feeRecipientForProposal resolves the fee recipient, warning once per proposer when there is none: +// building with the zero address gives that block's fees away. +func (a *ApiHandler) feeRecipientForProposal(proposerIndex, targetSlot uint64) common.Address { + feeRecipient, registered := a.validatorParams.GetFeeRecipient(proposerIndex) + if registered { + return feeRecipient + } + // Claimed in one step, so requests for the same slot arriving together do not each decide they + // are the first. Without somewhere to remember them, reporting every time beats reporting never. + firstTime := true + if a.unregisteredProposers != nil { + alreadyWarned, _ := a.unregisteredProposers.ContainsOrAdd(proposerIndex, struct{}{}) + firstTime = !alreadyWarned + } + if firstTime { + log.Warn("BlockProduction: no fee recipient from prepare_beacon_proposer, using zero address", + "proposerIndex", proposerIndex, "slot", targetSlot) + } + return common.Address{} +} + func shouldRetryGetPayload(now, deadline time.Time) bool { return now.Before(deadline) } @@ -297,6 +318,7 @@ func shouldRetryGetPayload(now, deadline time.Time) bool { // it returns a payload or the deadline passes; ok is false when no payload was produced in time. func pollAssembledPayload( ctx context.Context, + targetSlot uint64, window blockBuilderWindow, retryTime time.Duration, get func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error), @@ -314,12 +336,36 @@ func pollAssembledPayload( defer deadlineTimer.Stop() retryTicker := time.NewTicker(retryTime) defer retryTicker.Stop() + + // Polling retries at the retry cadence, so one failing slot reported once per attempt buried + // everything else in it. Report when the window is over, and only if it ended without a + // payload: an attempt that fails and then succeeds is a healthy slot. + var ( + failures int + firstErr error + collected bool + ) + defer func() { + if collected || failures == 0 { + return + } + log.Error("BlockProduction: failed to get payload", "slot", targetSlot, "failures", failures, "err", firstErr) + }() for { // Grab at least once, even past the deadline, so a late produce request still gets a payload. payload, bundles, requestsBundle, blockValue, err := get() if err != nil { - log.Error("BlockProduction: Failed to get payload", "err", err) + // The caller's own cancellation comes back through get. That is the slot ending, not + // the execution layer failing, and it is not worth reporting on a healthy node. A + // failure that happened before it still is. + if ctx.Err() == nil || !errors.Is(err, ctx.Err()) { + failures++ + if firstErr == nil { + firstErr = err + } + } } else if payload != nil { + collected = true return payload, bundles, requestsBundle, blockValue, true } select { @@ -1037,7 +1083,7 @@ func (a *ApiHandler) produceBeaconBody( log.Info("BlockProduction: ForkChoiceUpdate&GetPayload took", "duration", time.Since(start)) }() retryTime := 10 * time.Millisecond - feeRecipient, _ := a.validatorParams.GetFeeRecipient(proposerIndex) + feeRecipient := a.feeRecipientForProposal(proposerIndex, targetSlot) withdrawals, err := a.expectedWithdrawals(baseState, gloasWithdrawalsState, stateVersion, targetSlot) if err != nil { log.Error("BlockProduction: GetExpectedWithdrawals failed", "err", err) @@ -1073,7 +1119,7 @@ func (a *ApiHandler) produceBeaconBody( } slotStart := a.ethClock.GetSlotTime(targetSlot) buildWindow := computeBlockBuilderWindow(builderStartedAt, slotStart, a.beaconChainCfg, stateVersion) - payload, bundles, requestsBundle, blockValue, ok := pollAssembledPayload(ctx, buildWindow, retryTime, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { + payload, bundles, requestsBundle, blockValue, ok := pollAssembledPayload(ctx, targetSlot, buildWindow, retryTime, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { return a.engine.GetAssembledBlock(ctx, idBytes, stateVersion) }) if !ok { diff --git a/cl/beacon/handler/block_production_test.go b/cl/beacon/handler/block_production_test.go index fddce8a9234..11f195d15c0 100644 --- a/cl/beacon/handler/block_production_test.go +++ b/cl/beacon/handler/block_production_test.go @@ -24,6 +24,8 @@ import ( "math/big" "net/http" "net/http/httptest" + "strings" + "sync" "testing" "time" @@ -37,7 +39,9 @@ import ( "github.com/erigontech/erigon/cl/cltypes" "github.com/erigontech/erigon/cl/cltypes/solid" "github.com/erigontech/erigon/cl/phase1/core/state" + "github.com/erigontech/erigon/cl/phase1/core/state/lru" "github.com/erigontech/erigon/cl/phase1/execution_client" + "github.com/erigontech/erigon/cl/validator/validator_params" "github.com/erigontech/erigon/common" "github.com/erigontech/erigon/common/hexutil" "github.com/erigontech/erigon/common/log/v3" @@ -387,7 +391,7 @@ func TestPollAssembledPayloadReturnsReadyPayload(t *testing.T) { window := blockBuilderWindow{firstGetAt: now.Add(-time.Millisecond), pollUntil: now.Add(time.Second)} want := &cltypes.Eth1Block{} calls := 0 - payload, _, _, _, ok := pollAssembledPayload(context.Background(), window, time.Millisecond, + payload, _, _, _, ok := pollAssembledPayload(context.Background(), 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ return want, nil, nil, nil, nil @@ -402,7 +406,7 @@ func TestPollAssembledPayloadRetriesWhileBusy(t *testing.T) { window := blockBuilderWindow{firstGetAt: now, pollUntil: now.Add(time.Second)} want := &cltypes.Eth1Block{} calls := 0 - payload, _, _, _, ok := pollAssembledPayload(context.Background(), window, time.Millisecond, + payload, _, _, _, ok := pollAssembledPayload(context.Background(), 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ if calls < 3 { @@ -420,7 +424,7 @@ func TestPollAssembledPayloadRetriesOnError(t *testing.T) { window := blockBuilderWindow{firstGetAt: now, pollUntil: now.Add(time.Second)} want := &cltypes.Eth1Block{} calls := 0 - payload, _, _, _, ok := pollAssembledPayload(context.Background(), window, time.Millisecond, + payload, _, _, _, ok := pollAssembledPayload(context.Background(), 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ if calls == 1 { @@ -437,7 +441,7 @@ func TestPollAssembledPayloadStopsAtDeadline(t *testing.T) { now := time.Now() window := blockBuilderWindow{firstGetAt: now, pollUntil: now.Add(50 * time.Millisecond)} calls := 0 - payload, _, _, _, ok := pollAssembledPayload(context.Background(), window, time.Millisecond, + payload, _, _, _, ok := pollAssembledPayload(context.Background(), 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ return nil, nil, nil, nil, nil @@ -451,7 +455,7 @@ func TestPollAssembledPayloadLateRequestGrabsOnce(t *testing.T) { past := time.Now().Add(-time.Second) window := blockBuilderWindow{firstGetAt: past, pollUntil: past} calls := 0 - _, _, _, _, ok := pollAssembledPayload(context.Background(), window, time.Millisecond, + _, _, _, _, ok := pollAssembledPayload(context.Background(), 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ return nil, nil, nil, nil, nil @@ -466,7 +470,7 @@ func TestPollAssembledPayloadReturnsOnContextCancel(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() calls := 0 - _, _, _, _, ok := pollAssembledPayload(ctx, window, time.Millisecond, + _, _, _, _, ok := pollAssembledPayload(ctx, 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ return nil, nil, nil, nil, nil @@ -829,3 +833,169 @@ func TestPayloadAttributesOmitFieldsTheChosenVersionCannotCarry(t *testing.T) { }) } } + +// syncedBuffer is a writer the log package can hand to several goroutines at once, which +// StreamHandler requires and a bare bytes.Buffer does not provide. +type syncedBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (s *syncedBuffer) Write(p []byte) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + return s.buf.Write(p) +} + +func (s *syncedBuffer) String() string { + s.mu.Lock() + defer s.mu.Unlock() + return s.buf.String() +} + +// captureProductionLogs redirects the root logger for one test and returns the block-production +// records it wrote. Other components in this package log to the same root, sometimes from +// goroutines that outlive their own test, so the records are filtered rather than read whole. +func captureProductionLogs(t *testing.T) func() string { + t.Helper() + output := &syncedBuffer{} + previous := log.Root().GetHandler() + log.Root().SetHandler(log.StreamHandler(output, log.LogfmtFormat())) + t.Cleanup(func() { log.Root().SetHandler(previous) }) + return func() string { + var ours []string + for line := range strings.SplitSeq(output.String(), "\n") { + if strings.Contains(line, "BlockProduction:") { + ours = append(ours, line) + } + } + return strings.Join(ours, "\n") + } +} + +func TestPollAssembledPayloadStaysQuietWhenAFailedPollRecovers(t *testing.T) { + logs := captureProductionLogs(t) + window := blockBuilderWindow{firstGetAt: time.Now(), pollUntil: time.Now().Add(time.Second)} + + calls := 0 + payload, _, _, _, ok := pollAssembledPayload(t.Context(), 10, window, time.Millisecond, + func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { + calls++ + if calls == 1 { + return nil, nil, nil, nil, errors.New("execution module is busy") + } + return &cltypes.Eth1Block{}, &engine_types.BlobsBundle{}, nil, big.NewInt(1), nil + }) + + require.True(t, ok) + require.NotNil(t, payload) + // Contention that clears is a healthy slot, so nothing may be reported at error level. + require.NotContains(t, logs(), "lvl=eror") +} + +func TestPollAssembledPayloadReportsAWindowThatNeverProducedOnce(t *testing.T) { + logs := captureProductionLogs(t) + window := blockBuilderWindow{firstGetAt: time.Now(), pollUntil: time.Now().Add(50 * time.Millisecond)} + + calls := 0 + _, _, _, _, ok := pollAssembledPayload(t.Context(), 10, window, time.Millisecond, + func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { + calls++ + return nil, nil, nil, nil, errors.New("boom") + }) + + require.False(t, ok) + require.NotZero(t, calls) + + // One record for the whole window, carrying the slot, how many attempts failed, and the first + // reason, which is the one that says what went wrong. + captured := logs() + require.Equal(t, 1, strings.Count(captured, "lvl=eror")) + require.Contains(t, captured, "slot=10") + require.Contains(t, captured, "failures=") + require.Contains(t, captured, "boom") +} + +func TestPollAssembledPayloadStaysQuietWhenTheCallerGoesAway(t *testing.T) { + logs := captureProductionLogs(t) + ctx, cancel := context.WithCancel(t.Context()) + window := blockBuilderWindow{firstGetAt: time.Now(), pollUntil: time.Now().Add(time.Minute)} + + _, _, _, _, ok := pollAssembledPayload(ctx, 10, window, time.Millisecond, + func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { + cancel() + return nil, nil, nil, nil, context.Canceled + }) + + // A validator client that times out, or a node shutting down, takes the slot with it. Nothing + // failed that anyone can act on. + require.False(t, ok) + require.NotContains(t, logs(), "lvl=eror") +} + +func TestPollAssembledPayloadStillReportsFailuresThatPrecededTheCancellation(t *testing.T) { + logs := captureProductionLogs(t) + ctx, cancel := context.WithCancel(t.Context()) + window := blockBuilderWindow{firstGetAt: time.Now(), pollUntil: time.Now().Add(time.Minute)} + + calls := 0 + _, _, _, _, ok := pollAssembledPayload(ctx, 10, window, time.Millisecond, + func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { + calls++ + if calls == 1 { + return nil, nil, nil, nil, errors.New("boom") + } + cancel() + return nil, nil, nil, nil, context.Canceled + }) + + // The client may well have given up because production was failing. Dropping the record + // because of how the slot ended would lose the only sign of it. + require.False(t, ok) + captured := logs() + require.Equal(t, 1, strings.Count(captured, "lvl=eror")) + require.Contains(t, captured, "boom") +} + +func TestFeeRecipientWarnsOncePerProposer(t *testing.T) { + logs := captureProductionLogs(t) + warned, err := lru.New[uint64, struct{}]("unregisteredProposers", 8) + require.NoError(t, err) + params := validator_params.NewValidatorParams() + a := &ApiHandler{validatorParams: params, unregisteredProposers: warned} + + registered := common.Address{0x11} + params.SetFeeRecipient(7, registered) + require.Equal(t, registered, a.feeRecipientForProposal(7, 1)) + require.NotContains(t, logs(), "lvl=warn", "a registered proposer must stay quiet") + + // Giving the fees away is worth saying, but only once: a chain whose validator never registers + // one would otherwise warn on every proposal. + require.Equal(t, common.Address{}, a.feeRecipientForProposal(9, 2)) + require.Equal(t, common.Address{}, a.feeRecipientForProposal(9, 3)) + require.Equal(t, 1, strings.Count(logs(), "lvl=warn")) + + require.Equal(t, common.Address{}, a.feeRecipientForProposal(10, 4)) + require.Equal(t, 2, strings.Count(logs(), "lvl=warn"), "a different proposer is worth saying again") + + // Alternating proposers must not each reset the other: 9 has already been reported. + require.Equal(t, common.Address{}, a.feeRecipientForProposal(9, 5)) + require.Equal(t, 2, strings.Count(logs(), "lvl=warn")) +} + +func TestFeeRecipientWarnsOncePerProposerUnderConcurrentRequests(t *testing.T) { + logs := captureProductionLogs(t) + warned, err := lru.New[uint64, struct{}]("unregisteredProposers", 8) + require.NoError(t, err) + a := &ApiHandler{validatorParams: validator_params.NewValidatorParams(), unregisteredProposers: warned} + + // Several block template requests for the same slot arrive together, and each would otherwise + // find the proposer absent and report it. + var wg sync.WaitGroup + for range 128 { + wg.Go(func() { a.feeRecipientForProposal(9, 2) }) + } + wg.Wait() + + require.Equal(t, 1, strings.Count(logs(), "lvl=warn")) +} diff --git a/cl/beacon/handler/handler.go b/cl/beacon/handler/handler.go index 1f15ca474cb..8f2c4da814c 100644 --- a/cl/beacon/handler/handler.go +++ b/cl/beacon/handler/handler.go @@ -107,7 +107,10 @@ type ApiHandler struct { logger log.Logger // Validator data structures - validatorParams *validator_params.ValidatorParams + validatorParams *validator_params.ValidatorParams + // unregisteredProposers remembers which proposers have already been warned about, so the + // warning is once per proposer rather than once per proposal. + unregisteredProposers *lru.Cache[uint64, struct{}] blobBundles *lru.Cache[common.Bytes48, BlobBundle] // Keep recent bundled blobs from the execution layer. engine execution_client.ExecutionEngine elClientVersion atomic.Pointer[engine_types.ClientVersionV1] // Cached execution client version for default graffiti. @@ -199,6 +202,10 @@ func NewApiHandler( blobSnapshots = caplinSnapshots } + unregisteredProposers, err := lru.New[uint64, struct{}]("unregisteredProposers", 1024) + if err != nil { + panic(err) + } slotWaitedForAttestationProduction, err := lru.New[uint64, struct{}]("slotWaitedForAttestationProduction", 1024) if err != nil { panic(err) @@ -227,6 +234,7 @@ func NewApiHandler( caplinStateSnapshots: caplinStateSnapshots, peerDas: peerDas, slotWaitedForAttestationProduction: slotWaitedForAttestationProduction, + unregisteredProposers: unregisteredProposers, randaoMixesPool: sync.Pool{New: func() any { return solid.NewHashVector(int(beaconChainConfig.EpochsPerHistoricalVector)) }}, From 8ac03529417cc7ab2f69e804bb1b89c231e90226 Mon Sep 17 00:00:00 2001 From: Oleksandr Lystopad Date: Mon, 17 Aug 2026 15:33:32 +0200 Subject: [PATCH 2/5] cl/beacon: do not start another collection for a caller that has gone The window's select can pick its timer or its ticker while cancellation is ready too, and the next pass then called the execution layer with a context that had already ended. That call takes the module's semaphore before it looks at a context, so it comes back as contention rather than cancellation, which counts as a real failure and produces the record a departed caller is not supposed to produce. Failures seen before the caller left are still counted, and still reported: giving up is often the consequence of them. --- cl/beacon/handler/block_production.go | 6 ++++++ cl/beacon/handler/block_production_test.go | 21 +++++++++++++++++++++ 2 files changed, 27 insertions(+) diff --git a/cl/beacon/handler/block_production.go b/cl/beacon/handler/block_production.go index 04f9c6d4d10..289fb7d0d0c 100644 --- a/cl/beacon/handler/block_production.go +++ b/cl/beacon/handler/block_production.go @@ -352,6 +352,12 @@ func pollAssembledPayload( log.Error("BlockProduction: failed to get payload", "slot", targetSlot, "failures", failures, "err", firstErr) }() for { + // Nothing is waiting for this payload any more, and starting another collection would stop + // the builder and could fail for reasons of its own - contention, most likely - which would + // then be reported against a slot nobody is waiting for. + if ctx.Err() != nil { + return nil, nil, nil, nil, false + } // Grab at least once, even past the deadline, so a late produce request still gets a payload. payload, bundles, requestsBundle, blockValue, err := get() if err != nil { diff --git a/cl/beacon/handler/block_production_test.go b/cl/beacon/handler/block_production_test.go index 11f195d15c0..d63f27e212b 100644 --- a/cl/beacon/handler/block_production_test.go +++ b/cl/beacon/handler/block_production_test.go @@ -999,3 +999,24 @@ func TestFeeRecipientWarnsOncePerProposerUnderConcurrentRequests(t *testing.T) { require.Equal(t, 1, strings.Count(logs(), "lvl=warn")) } + +func TestPollAssembledPayloadDoesNotCollectAfterTheCallerHasGone(t *testing.T) { + logs := captureProductionLogs(t) + ctx, cancel := context.WithCancel(t.Context()) + cancel() + past := time.Now().Add(-time.Second) + window := blockBuilderWindow{firstGetAt: past, pollUntil: past} + + calls := 0 + _, _, _, _, ok := pollAssembledPayload(ctx, 10, window, time.Microsecond, + func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { + calls++ + // The execution module takes its semaphore before it looks at a context, so a request + // made after the caller has gone comes back as contention rather than cancellation. + return nil, nil, nil, nil, errors.New("execution module is busy") + }) + + require.False(t, ok) + require.Zero(t, calls, "collection must not be started for a caller that has gone") + require.NotContains(t, logs(), "lvl=eror") +} From 4b58bb6a730c78c12c75b1e1fc656e6d19cfa83d Mon Sep 17 00:00:00 2001 From: Oleksandr Lystopad Date: Mon, 17 Aug 2026 16:20:40 +0200 Subject: [PATCH 3/5] cl/beacon: let one boundary own the record for a failed production Reporting once inside the polling loop was not enough: the loop returned only whether it had a payload, so the layers above it invented a cause-less failure of their own, produceBlock logged that, and the endpoint logged it again. A validator client disconnecting produced an error about a healthy node, and a real polling failure produced several records, the useful one buried among them. Every failure in the body goroutine now carries its cause out, all ten of them rather than the few that already did, and produceBlock reports once at its error boundary. A caller that has already gone is not raised there; an execution layer that stopped answering still is, since that arrives as a deadline rather than a cancellation. The polling loop keeps its own accounting only to decide what to hand back: a caller that left with nothing having failed took the slot with it, while anything that failed before that is worth reporting however the window ended. The log capture in tests no longer filters by message. Filtering to this file's own prefix is what hid the outer records, so the tests agreed with a claim that was not true. --- cl/beacon/handler/block_production.go | 96 +++++++++------ cl/beacon/handler/block_production_test.go | 129 ++++++++++++++------- 2 files changed, 147 insertions(+), 78 deletions(-) diff --git a/cl/beacon/handler/block_production.go b/cl/beacon/handler/block_production.go index 289fb7d0d0c..1ddfcab34f7 100644 --- a/cl/beacon/handler/block_production.go +++ b/cl/beacon/handler/block_production.go @@ -310,6 +310,19 @@ func (a *ApiHandler) feeRecipientForProposal(proposerIndex, targetSlot uint64) c return common.Address{} } +// reportProductionFailure is the one place a failed production is recorded, so every exit through +// produceBlock is covered exactly once. A caller that has already gone is not something anyone can +// act on; an execution layer that stopped answering still is, and that arrives as a deadline. +func reportProductionFailure(err error, targetSlot uint64) { + switch { + case err == nil: + case errors.Is(err, context.Canceled): + log.Debug("BlockProduction: abandoned by its caller", "err", err, "slot", targetSlot) + default: + log.Error("BlockProduction: failed to produce block", "err", err, "slot", targetSlot) + } +} + func shouldRetryGetPayload(now, deadline time.Time) bool { return now.Before(deadline) } @@ -322,13 +335,13 @@ func pollAssembledPayload( window blockBuilderWindow, retryTime time.Duration, get func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error), -) (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, bool) { +) (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { if wait := time.Until(window.firstGetAt); wait > 0 { buildTimer := time.NewTimer(wait) select { case <-ctx.Done(): buildTimer.Stop() - return nil, nil, nil, nil, false + return nil, nil, nil, nil, ctx.Err() case <-buildTimer.C: } } @@ -337,29 +350,21 @@ func pollAssembledPayload( retryTicker := time.NewTicker(retryTime) defer retryTicker.Stop() - // Polling retries at the retry cadence, so one failing slot reported once per attempt buried - // everything else in it. Report when the window is over, and only if it ended without a - // payload: an attempt that fails and then succeeds is a healthy slot. var ( - failures int - firstErr error - collected bool + attempts int + failures int + firstErr error ) - defer func() { - if collected || failures == 0 { - return - } - log.Error("BlockProduction: failed to get payload", "slot", targetSlot, "failures", failures, "err", firstErr) - }() for { // Nothing is waiting for this payload any more, and starting another collection would stop // the builder and could fail for reasons of its own - contention, most likely - which would // then be reported against a slot nobody is waiting for. if ctx.Err() != nil { - return nil, nil, nil, nil, false + return nil, nil, nil, nil, terminalCause(ctx, attempts, failures, firstErr) } // Grab at least once, even past the deadline, so a late produce request still gets a payload. payload, bundles, requestsBundle, blockValue, err := get() + attempts++ if err != nil { // The caller's own cancellation comes back through get. That is the slot ending, not // the execution layer failing, and it is not worth reporting on a healthy node. A @@ -371,24 +376,43 @@ func pollAssembledPayload( } } } else if payload != nil { - collected = true - return payload, bundles, requestsBundle, blockValue, true + return payload, bundles, requestsBundle, blockValue, nil } select { case <-ctx.Done(): - return nil, nil, nil, nil, false + return nil, nil, nil, nil, terminalCause(ctx, attempts, failures, firstErr) case <-deadlineTimer.C: - return nil, nil, nil, nil, false + return nil, nil, nil, nil, terminalCause(ctx, attempts, failures, firstErr) case <-retryTicker.C: } // Re-check here, not before get(): the select may pick the ticker after // deadlineTimer fired, and get() stops the builder, so it must not run past pollUntil. if !shouldRetryGetPayload(time.Now(), window.pollUntil) { - return nil, nil, nil, nil, false + return nil, nil, nil, nil, terminalCause(ctx, attempts, failures, firstErr) } } } +// terminalCause says why the window ended without a payload. A caller that went away with nothing +// having failed took the slot with it, and there is nothing anyone can act on; anything that did +// fail is worth reporting however the window ended, since giving up is often the consequence of it. +func terminalCause(ctx context.Context, attempts, failures int, firstErr error) error { + if failures == 0 && ctx.Err() != nil { + return ctx.Err() + } + if firstErr != nil { + return fmt.Errorf("no payload after %s, %d failed: %w", attemptsMade(attempts), failures, firstErr) + } + return fmt.Errorf("no payload after %s", attemptsMade(attempts)) +} + +func attemptsMade(attempts int) string { + if attempts == 1 { + return "1 attempt" + } + return fmt.Sprintf("%d attempts", attempts) +} + func (a *ApiHandler) waitForHeadSlot(slot uint64) { stopCh := time.After(time.Second) for { @@ -611,7 +635,7 @@ func (a *ApiHandler) GetEthV3ValidatorBlock( log.Info("[Beacon API] Found BeaconState object for block production", "slot", targetSlot, "duration", time.Since(start)) block, err := a.produceBlock(ctx, builderBoostFactor, baseBlockSlot, baseBlockRoot, baseState, targetSlot, randaoReveal, graffiti) if err != nil { - log.Warn("Failed to produce block", "err", err, "slot", targetSlot) + // produceBlock owns this record; repeating it here made one failure two. return nil, err } @@ -729,7 +753,9 @@ func (a *ApiHandler) produceBlock( targetSlot uint64, randaoReveal common.Bytes96, graffiti common.Hash, -) (*cltypes.BlindOrExecutionBeaconBlock, error) { +) (block *cltypes.BlindOrExecutionBeaconBlock, err error) { + defer func() { reportProductionFailure(err, targetSlot) }() + var wg sync.WaitGroup // produce beacon body var ( @@ -795,7 +821,6 @@ func (a *ApiHandler) produceBlock( if localErr != nil { // if we failed to locally produce the beacon body, we should not proceed with the block production - log.Error("Failed to produce beacon body", "err", localErr, "slot", targetSlot) return nil, localErr } // prepare basic block @@ -807,7 +832,7 @@ func (a *ApiHandler) produceBlock( if err != nil { return nil, err } - block := &cltypes.BlindOrExecutionBeaconBlock{ + block = &cltypes.BlindOrExecutionBeaconBlock{ Slot: targetSlot, ProposerIndex: proposerIndex, ParentRoot: baseBlockRoot, @@ -1092,7 +1117,7 @@ func (a *ApiHandler) produceBeaconBody( feeRecipient := a.feeRecipientForProposal(proposerIndex, targetSlot) withdrawals, err := a.expectedWithdrawals(baseState, gloasWithdrawalsState, stateVersion, targetSlot) if err != nil { - log.Error("BlockProduction: GetExpectedWithdrawals failed", "err", err) + executionErr = fmt.Errorf("produceBeaconBody: expected withdrawals: %w", err) return } slotNumber := hexutil.Uint64(targetSlot) @@ -1116,7 +1141,7 @@ func (a *ApiHandler) produceBeaconBody( stateVersion, ) if err != nil { - log.Error("BlockProduction: Failed to get payload id", "err", err) + executionErr = fmt.Errorf("produceBeaconBody: forkchoice update: %w", err) return } if len(idBytes) == 0 { @@ -1125,10 +1150,11 @@ func (a *ApiHandler) produceBeaconBody( } slotStart := a.ethClock.GetSlotTime(targetSlot) buildWindow := computeBlockBuilderWindow(builderStartedAt, slotStart, a.beaconChainCfg, stateVersion) - payload, bundles, requestsBundle, blockValue, ok := pollAssembledPayload(ctx, targetSlot, buildWindow, retryTime, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { + payload, bundles, requestsBundle, blockValue, pollErr := pollAssembledPayload(ctx, targetSlot, buildWindow, retryTime, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { return a.engine.GetAssembledBlock(ctx, idBytes, stateVersion) }) - if !ok { + if pollErr != nil { + executionErr = fmt.Errorf("produceBeaconBody: %w", pollErr) return } // Determine block value @@ -1141,28 +1167,28 @@ func (a *ApiHandler) produceBeaconBody( if stateVersion.Before(clparams.FuluVersion) { if len(bundles.Blobs) != len(bundles.Proofs) || len(bundles.Commitments) != len(bundles.Proofs) { - log.Error("BlockProduction: Invalid bundle") + executionErr = errors.New("produceBeaconBody: invalid blobs bundle") return } } else { if len(bundles.Blobs) != len(bundles.Commitments) || len(bundles.Proofs) != len(bundles.Blobs)*int(a.beaconChainCfg.NumberOfColumns) { - log.Error("BlockProduction: Invalid peerdas bundle") + executionErr = errors.New("produceBeaconBody: invalid peerdas bundle") return } } for i := range bundles.Blobs { if len(bundles.Commitments[i]) != length.Bytes48 { - log.Error("BlockProduction: Invalid commitment length") + executionErr = errors.New("produceBeaconBody: invalid commitment length") return } if stateVersion.Before(clparams.FuluVersion) && len(bundles.Proofs[i]) != length.Bytes48 { - log.Error("BlockProduction: Invalid proof length") + executionErr = errors.New("produceBeaconBody: invalid proof length") return } if len(bundles.Blobs[i]) != cltypes.BYTES_PER_BLOB { - log.Error("BlockProduction: Invalid blob length") + executionErr = errors.New("produceBeaconBody: invalid blob length") return } @@ -1229,7 +1255,7 @@ func (a *ApiHandler) produceBeaconBody( // so the bid's ExecutionRequestsRoot matches the envelope's actual root. root, err := gloasExecRequests.HashSSZ() if err != nil { - log.Error("BlockProduction: GLOAS failed to compute ExecutionRequestsRoot", "err", err) + executionErr = fmt.Errorf("produceBeaconBody: execution requests root: %w", err) } else { executionRequestsRoot = common.Hash(root) } @@ -1279,7 +1305,7 @@ func (a *ApiHandler) produceBeaconBody( }() beaconBody.SyncAggregate, err = a.syncMessagePool.GetSyncAggregate(targetSlot-1, blockRoot) if err != nil { - log.Error("BlockProduction: Failed to get sync aggregate", "err", err) + executionErr = fmt.Errorf("produceBeaconBody: sync aggregate: %w", err) } }) // Process operations all in parallel with each other. diff --git a/cl/beacon/handler/block_production_test.go b/cl/beacon/handler/block_production_test.go index d63f27e212b..a85dc77beb9 100644 --- a/cl/beacon/handler/block_production_test.go +++ b/cl/beacon/handler/block_production_test.go @@ -391,12 +391,12 @@ func TestPollAssembledPayloadReturnsReadyPayload(t *testing.T) { window := blockBuilderWindow{firstGetAt: now.Add(-time.Millisecond), pollUntil: now.Add(time.Second)} want := &cltypes.Eth1Block{} calls := 0 - payload, _, _, _, ok := pollAssembledPayload(context.Background(), 10, window, time.Millisecond, + payload, _, _, _, pollErr := pollAssembledPayload(t.Context(), 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ return want, nil, nil, nil, nil }) - require.True(t, ok) + require.NoError(t, pollErr) require.Same(t, want, payload) require.Equal(t, 1, calls) } @@ -406,7 +406,7 @@ func TestPollAssembledPayloadRetriesWhileBusy(t *testing.T) { window := blockBuilderWindow{firstGetAt: now, pollUntil: now.Add(time.Second)} want := &cltypes.Eth1Block{} calls := 0 - payload, _, _, _, ok := pollAssembledPayload(context.Background(), 10, window, time.Millisecond, + payload, _, _, _, pollErr := pollAssembledPayload(t.Context(), 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ if calls < 3 { @@ -414,7 +414,7 @@ func TestPollAssembledPayloadRetriesWhileBusy(t *testing.T) { } return want, nil, nil, nil, nil }) - require.True(t, ok) + require.NoError(t, pollErr) require.Same(t, want, payload) require.Equal(t, 3, calls) } @@ -424,7 +424,7 @@ func TestPollAssembledPayloadRetriesOnError(t *testing.T) { window := blockBuilderWindow{firstGetAt: now, pollUntil: now.Add(time.Second)} want := &cltypes.Eth1Block{} calls := 0 - payload, _, _, _, ok := pollAssembledPayload(context.Background(), 10, window, time.Millisecond, + payload, _, _, _, pollErr := pollAssembledPayload(t.Context(), 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ if calls == 1 { @@ -432,7 +432,7 @@ func TestPollAssembledPayloadRetriesOnError(t *testing.T) { } return want, nil, nil, nil, nil }) - require.True(t, ok) + require.NoError(t, pollErr) require.Same(t, want, payload) require.Equal(t, 2, calls) } @@ -441,12 +441,12 @@ func TestPollAssembledPayloadStopsAtDeadline(t *testing.T) { now := time.Now() window := blockBuilderWindow{firstGetAt: now, pollUntil: now.Add(50 * time.Millisecond)} calls := 0 - payload, _, _, _, ok := pollAssembledPayload(context.Background(), 10, window, time.Millisecond, + payload, _, _, _, pollErr := pollAssembledPayload(t.Context(), 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ return nil, nil, nil, nil, nil }) - require.False(t, ok) + require.Error(t, pollErr) require.Nil(t, payload) require.NotZero(t, calls) } @@ -455,12 +455,12 @@ func TestPollAssembledPayloadLateRequestGrabsOnce(t *testing.T) { past := time.Now().Add(-time.Second) window := blockBuilderWindow{firstGetAt: past, pollUntil: past} calls := 0 - _, _, _, _, ok := pollAssembledPayload(context.Background(), 10, window, time.Millisecond, + _, _, _, _, pollErr := pollAssembledPayload(t.Context(), 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ return nil, nil, nil, nil, nil }) - require.False(t, ok) + require.Error(t, pollErr) require.Equal(t, 1, calls) } @@ -470,12 +470,12 @@ func TestPollAssembledPayloadReturnsOnContextCancel(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() calls := 0 - _, _, _, _, ok := pollAssembledPayload(ctx, 10, window, time.Millisecond, + _, _, _, _, pollErr := pollAssembledPayload(ctx, 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ return nil, nil, nil, nil, nil }) - require.False(t, ok) + require.Error(t, pollErr) require.Zero(t, calls) } @@ -853,9 +853,9 @@ func (s *syncedBuffer) String() string { return s.buf.String() } -// captureProductionLogs redirects the root logger for one test and returns the block-production -// records it wrote. Other components in this package log to the same root, sometimes from -// goroutines that outlive their own test, so the records are filtered rather than read whole. +// captureProductionLogs redirects the root logger for one test and returns everything written at +// warning level or above. It deliberately does not filter by message: a record this package emits +// under another name is exactly what a test asserting silence needs to see. func captureProductionLogs(t *testing.T) func() string { t.Helper() output := &syncedBuffer{} @@ -863,13 +863,13 @@ func captureProductionLogs(t *testing.T) func() string { log.Root().SetHandler(log.StreamHandler(output, log.LogfmtFormat())) t.Cleanup(func() { log.Root().SetHandler(previous) }) return func() string { - var ours []string + var loud []string for line := range strings.SplitSeq(output.String(), "\n") { - if strings.Contains(line, "BlockProduction:") { - ours = append(ours, line) + if strings.Contains(line, "lvl=eror") || strings.Contains(line, "lvl=warn") { + loud = append(loud, line) } } - return strings.Join(ours, "\n") + return strings.Join(loud, "\n") } } @@ -878,7 +878,7 @@ func TestPollAssembledPayloadStaysQuietWhenAFailedPollRecovers(t *testing.T) { window := blockBuilderWindow{firstGetAt: time.Now(), pollUntil: time.Now().Add(time.Second)} calls := 0 - payload, _, _, _, ok := pollAssembledPayload(t.Context(), 10, window, time.Millisecond, + payload, _, _, _, pollErr := pollAssembledPayload(t.Context(), 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ if calls == 1 { @@ -887,7 +887,7 @@ func TestPollAssembledPayloadStaysQuietWhenAFailedPollRecovers(t *testing.T) { return &cltypes.Eth1Block{}, &engine_types.BlobsBundle{}, nil, big.NewInt(1), nil }) - require.True(t, ok) + require.NoError(t, pollErr) require.NotNil(t, payload) // Contention that clears is a healthy slot, so nothing may be reported at error level. require.NotContains(t, logs(), "lvl=eror") @@ -897,23 +897,21 @@ func TestPollAssembledPayloadReportsAWindowThatNeverProducedOnce(t *testing.T) { logs := captureProductionLogs(t) window := blockBuilderWindow{firstGetAt: time.Now(), pollUntil: time.Now().Add(50 * time.Millisecond)} + boom := errors.New("boom") calls := 0 - _, _, _, _, ok := pollAssembledPayload(t.Context(), 10, window, time.Millisecond, + _, _, _, _, pollErr := pollAssembledPayload(t.Context(), 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ - return nil, nil, nil, nil, errors.New("boom") + return nil, nil, nil, nil, boom }) - require.False(t, ok) require.NotZero(t, calls) - // One record for the whole window, carrying the slot, how many attempts failed, and the first - // reason, which is the one that says what went wrong. - captured := logs() - require.Equal(t, 1, strings.Count(captured, "lvl=eror")) - require.Contains(t, captured, "slot=10") - require.Contains(t, captured, "failures=") - require.Contains(t, captured, "boom") + // The reason goes to the caller, which knows the slot and owns the record, carrying the first + // failure - the one that says what went wrong - and how many there were. + require.ErrorIs(t, pollErr, boom) + require.Contains(t, pollErr.Error(), "attempt") + require.Empty(t, logs(), "the poll does not report; its caller does") } func TestPollAssembledPayloadStaysQuietWhenTheCallerGoesAway(t *testing.T) { @@ -921,7 +919,7 @@ func TestPollAssembledPayloadStaysQuietWhenTheCallerGoesAway(t *testing.T) { ctx, cancel := context.WithCancel(t.Context()) window := blockBuilderWindow{firstGetAt: time.Now(), pollUntil: time.Now().Add(time.Minute)} - _, _, _, _, ok := pollAssembledPayload(ctx, 10, window, time.Millisecond, + _, _, _, _, pollErr := pollAssembledPayload(ctx, 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { cancel() return nil, nil, nil, nil, context.Canceled @@ -929,17 +927,16 @@ func TestPollAssembledPayloadStaysQuietWhenTheCallerGoesAway(t *testing.T) { // A validator client that times out, or a node shutting down, takes the slot with it. Nothing // failed that anyone can act on. - require.False(t, ok) + require.Error(t, pollErr) require.NotContains(t, logs(), "lvl=eror") } func TestPollAssembledPayloadStillReportsFailuresThatPrecededTheCancellation(t *testing.T) { - logs := captureProductionLogs(t) ctx, cancel := context.WithCancel(t.Context()) window := blockBuilderWindow{firstGetAt: time.Now(), pollUntil: time.Now().Add(time.Minute)} calls := 0 - _, _, _, _, ok := pollAssembledPayload(ctx, 10, window, time.Millisecond, + _, _, _, _, pollErr := pollAssembledPayload(ctx, 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ if calls == 1 { @@ -949,12 +946,10 @@ func TestPollAssembledPayloadStillReportsFailuresThatPrecededTheCancellation(t * return nil, nil, nil, nil, context.Canceled }) - // The client may well have given up because production was failing. Dropping the record - // because of how the slot ended would lose the only sign of it. - require.False(t, ok) - captured := logs() - require.Equal(t, 1, strings.Count(captured, "lvl=eror")) - require.Contains(t, captured, "boom") + // The client may well have given up because production was failing. Reporting only the + // cancellation would lose the only sign of it. + require.NotErrorIs(t, pollErr, context.Canceled) + require.Contains(t, pollErr.Error(), "boom") } func TestFeeRecipientWarnsOncePerProposer(t *testing.T) { @@ -1008,7 +1003,7 @@ func TestPollAssembledPayloadDoesNotCollectAfterTheCallerHasGone(t *testing.T) { window := blockBuilderWindow{firstGetAt: past, pollUntil: past} calls := 0 - _, _, _, _, ok := pollAssembledPayload(ctx, 10, window, time.Microsecond, + _, _, _, _, pollErr := pollAssembledPayload(ctx, 10, window, time.Microsecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ // The execution module takes its semaphore before it looks at a context, so a request @@ -1016,7 +1011,55 @@ func TestPollAssembledPayloadDoesNotCollectAfterTheCallerHasGone(t *testing.T) { return nil, nil, nil, nil, errors.New("execution module is busy") }) - require.False(t, ok) + require.Error(t, pollErr) require.Zero(t, calls, "collection must not be started for a caller that has gone") require.NotContains(t, logs(), "lvl=eror") } + +// produceBlockWithFailingCollection drives a real production through to the payload collection and +// makes that collection fail the given way, so the records the whole request emits are observable +// rather than only those of the polling loop. +func produceBlockWithFailingCollection(t *testing.T, ctx context.Context, collect error) error { + t.Helper() + ctrl := gomock.NewController(t) + _, _, _, _, postState, handler, _, _, _, _ := setupTestingHandler(t, clparams.ElectraVersion, log.Root(), false) + + engine := execution_client.NewMockExecutionEngine(ctrl) + engine.EXPECT().ForkChoiceUpdate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()). + Return([]byte{1, 2, 3, 4, 5, 6, 7, 8}, nil).AnyTimes() + engine.EXPECT().GetAssembledBlock(gomock.Any(), gomock.Any(), gomock.Any()). + Return(nil, nil, nil, nil, collect).AnyTimes() + engine.EXPECT().SupportInsertion().Return(true).AnyTimes() + handler.engine = engine + + _, err := handler.produceBlock(ctx, 1, postState.Slot(), common.Hash{0x41}, postState, + postState.Slot()+1, common.Bytes96{}, common.Hash{}) + return err +} + +func TestProductionReportsAFailedCollectionExactlyOnce(t *testing.T) { + logs := captureProductionLogs(t) + + err := produceBlockWithFailingCollection(t, t.Context(), errors.New("boom")) + require.Error(t, err) + + // One record for the whole request, and it carries the cause: the generic failure the caller + // used to see said only that production failed. + captured := logs() + require.Equal(t, 1, strings.Count(captured, "lvl=eror"), "records:\n"+captured) + require.Contains(t, captured, "boom") +} + +func TestProductionSaysNothingWhenTheRequestWasAbandoned(t *testing.T) { + logs := captureProductionLogs(t) + ctx, cancel := context.WithCancel(t.Context()) + cancel() + + err := produceBlockWithFailingCollection(t, ctx, context.Canceled) + require.Error(t, err) + + // A validator client that disconnects, or a node shutting down, is routine. Nothing about it is + // actionable, at any layer. The unregistered fee recipient this fixture also warns about is a + // separate matter and not what this measures. + require.NotContains(t, logs(), "lvl=eror", "records:\n"+logs()) +} From 9015ef96dd566097a1e4e80edaec01d1c78638ca Mon Sep 17 00:00:00 2001 From: Oleksandr Lystopad Date: Tue, 18 Aug 2026 10:00:30 +0200 Subject: [PATCH 4/5] cl/beacon: use the plain err name and one ctx per test The poll tests carried a pollErr name from when the function returned a bool, and took t.Context() inline at each call site. Both now match the rest of the file. --- cl/beacon/handler/block_production_test.go | 58 ++++++++++++---------- 1 file changed, 33 insertions(+), 25 deletions(-) diff --git a/cl/beacon/handler/block_production_test.go b/cl/beacon/handler/block_production_test.go index a85dc77beb9..0f0d08d47d4 100644 --- a/cl/beacon/handler/block_production_test.go +++ b/cl/beacon/handler/block_production_test.go @@ -387,26 +387,28 @@ func TestShouldRetryGetPayloadStopsAtDeadline(t *testing.T) { } func TestPollAssembledPayloadReturnsReadyPayload(t *testing.T) { + ctx := t.Context() now := time.Now() window := blockBuilderWindow{firstGetAt: now.Add(-time.Millisecond), pollUntil: now.Add(time.Second)} want := &cltypes.Eth1Block{} calls := 0 - payload, _, _, _, pollErr := pollAssembledPayload(t.Context(), 10, window, time.Millisecond, + payload, _, _, _, err := pollAssembledPayload(ctx, 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ return want, nil, nil, nil, nil }) - require.NoError(t, pollErr) + require.NoError(t, err) require.Same(t, want, payload) require.Equal(t, 1, calls) } func TestPollAssembledPayloadRetriesWhileBusy(t *testing.T) { + ctx := t.Context() now := time.Now() window := blockBuilderWindow{firstGetAt: now, pollUntil: now.Add(time.Second)} want := &cltypes.Eth1Block{} calls := 0 - payload, _, _, _, pollErr := pollAssembledPayload(t.Context(), 10, window, time.Millisecond, + payload, _, _, _, err := pollAssembledPayload(ctx, 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ if calls < 3 { @@ -414,17 +416,18 @@ func TestPollAssembledPayloadRetriesWhileBusy(t *testing.T) { } return want, nil, nil, nil, nil }) - require.NoError(t, pollErr) + require.NoError(t, err) require.Same(t, want, payload) require.Equal(t, 3, calls) } func TestPollAssembledPayloadRetriesOnError(t *testing.T) { + ctx := t.Context() now := time.Now() window := blockBuilderWindow{firstGetAt: now, pollUntil: now.Add(time.Second)} want := &cltypes.Eth1Block{} calls := 0 - payload, _, _, _, pollErr := pollAssembledPayload(t.Context(), 10, window, time.Millisecond, + payload, _, _, _, err := pollAssembledPayload(ctx, 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ if calls == 1 { @@ -432,35 +435,37 @@ func TestPollAssembledPayloadRetriesOnError(t *testing.T) { } return want, nil, nil, nil, nil }) - require.NoError(t, pollErr) + require.NoError(t, err) require.Same(t, want, payload) require.Equal(t, 2, calls) } func TestPollAssembledPayloadStopsAtDeadline(t *testing.T) { + ctx := t.Context() now := time.Now() window := blockBuilderWindow{firstGetAt: now, pollUntil: now.Add(50 * time.Millisecond)} calls := 0 - payload, _, _, _, pollErr := pollAssembledPayload(t.Context(), 10, window, time.Millisecond, + payload, _, _, _, err := pollAssembledPayload(ctx, 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ return nil, nil, nil, nil, nil }) - require.Error(t, pollErr) + require.Error(t, err) require.Nil(t, payload) require.NotZero(t, calls) } func TestPollAssembledPayloadLateRequestGrabsOnce(t *testing.T) { + ctx := t.Context() past := time.Now().Add(-time.Second) window := blockBuilderWindow{firstGetAt: past, pollUntil: past} calls := 0 - _, _, _, _, pollErr := pollAssembledPayload(t.Context(), 10, window, time.Millisecond, + _, _, _, _, err := pollAssembledPayload(ctx, 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ return nil, nil, nil, nil, nil }) - require.Error(t, pollErr) + require.Error(t, err) require.Equal(t, 1, calls) } @@ -470,12 +475,12 @@ func TestPollAssembledPayloadReturnsOnContextCancel(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() calls := 0 - _, _, _, _, pollErr := pollAssembledPayload(ctx, 10, window, time.Millisecond, + _, _, _, _, err := pollAssembledPayload(ctx, 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ return nil, nil, nil, nil, nil }) - require.Error(t, pollErr) + require.Error(t, err) require.Zero(t, calls) } @@ -874,11 +879,12 @@ func captureProductionLogs(t *testing.T) func() string { } func TestPollAssembledPayloadStaysQuietWhenAFailedPollRecovers(t *testing.T) { + ctx := t.Context() logs := captureProductionLogs(t) window := blockBuilderWindow{firstGetAt: time.Now(), pollUntil: time.Now().Add(time.Second)} calls := 0 - payload, _, _, _, pollErr := pollAssembledPayload(t.Context(), 10, window, time.Millisecond, + payload, _, _, _, err := pollAssembledPayload(ctx, 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ if calls == 1 { @@ -887,19 +893,20 @@ func TestPollAssembledPayloadStaysQuietWhenAFailedPollRecovers(t *testing.T) { return &cltypes.Eth1Block{}, &engine_types.BlobsBundle{}, nil, big.NewInt(1), nil }) - require.NoError(t, pollErr) + require.NoError(t, err) require.NotNil(t, payload) // Contention that clears is a healthy slot, so nothing may be reported at error level. require.NotContains(t, logs(), "lvl=eror") } func TestPollAssembledPayloadReportsAWindowThatNeverProducedOnce(t *testing.T) { + ctx := t.Context() logs := captureProductionLogs(t) window := blockBuilderWindow{firstGetAt: time.Now(), pollUntil: time.Now().Add(50 * time.Millisecond)} boom := errors.New("boom") calls := 0 - _, _, _, _, pollErr := pollAssembledPayload(t.Context(), 10, window, time.Millisecond, + _, _, _, _, err := pollAssembledPayload(ctx, 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ return nil, nil, nil, nil, boom @@ -909,8 +916,8 @@ func TestPollAssembledPayloadReportsAWindowThatNeverProducedOnce(t *testing.T) { // The reason goes to the caller, which knows the slot and owns the record, carrying the first // failure - the one that says what went wrong - and how many there were. - require.ErrorIs(t, pollErr, boom) - require.Contains(t, pollErr.Error(), "attempt") + require.ErrorIs(t, err, boom) + require.Contains(t, err.Error(), "attempt") require.Empty(t, logs(), "the poll does not report; its caller does") } @@ -919,7 +926,7 @@ func TestPollAssembledPayloadStaysQuietWhenTheCallerGoesAway(t *testing.T) { ctx, cancel := context.WithCancel(t.Context()) window := blockBuilderWindow{firstGetAt: time.Now(), pollUntil: time.Now().Add(time.Minute)} - _, _, _, _, pollErr := pollAssembledPayload(ctx, 10, window, time.Millisecond, + _, _, _, _, err := pollAssembledPayload(ctx, 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { cancel() return nil, nil, nil, nil, context.Canceled @@ -927,7 +934,7 @@ func TestPollAssembledPayloadStaysQuietWhenTheCallerGoesAway(t *testing.T) { // A validator client that times out, or a node shutting down, takes the slot with it. Nothing // failed that anyone can act on. - require.Error(t, pollErr) + require.Error(t, err) require.NotContains(t, logs(), "lvl=eror") } @@ -936,7 +943,7 @@ func TestPollAssembledPayloadStillReportsFailuresThatPrecededTheCancellation(t * window := blockBuilderWindow{firstGetAt: time.Now(), pollUntil: time.Now().Add(time.Minute)} calls := 0 - _, _, _, _, pollErr := pollAssembledPayload(ctx, 10, window, time.Millisecond, + _, _, _, _, err := pollAssembledPayload(ctx, 10, window, time.Millisecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ if calls == 1 { @@ -948,8 +955,8 @@ func TestPollAssembledPayloadStillReportsFailuresThatPrecededTheCancellation(t * // The client may well have given up because production was failing. Reporting only the // cancellation would lose the only sign of it. - require.NotErrorIs(t, pollErr, context.Canceled) - require.Contains(t, pollErr.Error(), "boom") + require.NotErrorIs(t, err, context.Canceled) + require.Contains(t, err.Error(), "boom") } func TestFeeRecipientWarnsOncePerProposer(t *testing.T) { @@ -1003,7 +1010,7 @@ func TestPollAssembledPayloadDoesNotCollectAfterTheCallerHasGone(t *testing.T) { window := blockBuilderWindow{firstGetAt: past, pollUntil: past} calls := 0 - _, _, _, _, pollErr := pollAssembledPayload(ctx, 10, window, time.Microsecond, + _, _, _, _, err := pollAssembledPayload(ctx, 10, window, time.Microsecond, func() (*cltypes.Eth1Block, *engine_types.BlobsBundle, *typesproto.RequestsBundle, *big.Int, error) { calls++ // The execution module takes its semaphore before it looks at a context, so a request @@ -1011,7 +1018,7 @@ func TestPollAssembledPayloadDoesNotCollectAfterTheCallerHasGone(t *testing.T) { return nil, nil, nil, nil, errors.New("execution module is busy") }) - require.Error(t, pollErr) + require.Error(t, err) require.Zero(t, calls, "collection must not be started for a caller that has gone") require.NotContains(t, logs(), "lvl=eror") } @@ -1038,9 +1045,10 @@ func produceBlockWithFailingCollection(t *testing.T, ctx context.Context, collec } func TestProductionReportsAFailedCollectionExactlyOnce(t *testing.T) { + ctx := t.Context() logs := captureProductionLogs(t) - err := produceBlockWithFailingCollection(t, t.Context(), errors.New("boom")) + err := produceBlockWithFailingCollection(t, ctx, errors.New("boom")) require.Error(t, err) // One record for the whole request, and it carries the cause: the generic failure the caller From 80d206574fd56b84f5af74dd1c566ef058dc964d Mon Sep 17 00:00:00 2001 From: Oleksandr Lystopad Date: Tue, 18 Aug 2026 10:12:42 +0200 Subject: [PATCH 5/5] cl/beacon: give each concurrent body step its own error to report The body steps run under one WaitGroup, so routing them all through a single executionErr was a write-write race as soon as two of them failed together. The sync aggregate now reports through its own variable, read after Wait like the others. It also stopped assigning to the enclosing err, which the surrounding goroutines have no reason to share. --- cl/beacon/handler/block_production.go | 13 +++++++++--- cl/beacon/handler/block_production_test.go | 24 ++++++++++++++++++++++ 2 files changed, 34 insertions(+), 3 deletions(-) diff --git a/cl/beacon/handler/block_production.go b/cl/beacon/handler/block_production.go index 1ddfcab34f7..4f4ba1405c8 100644 --- a/cl/beacon/handler/block_production.go +++ b/cl/beacon/handler/block_production.go @@ -1097,7 +1097,9 @@ func (a *ApiHandler) produceBeaconBody( var executionPayload *cltypes.Eth1Block var executionValue uint64 - var executionErr error + // One collector per concurrent body step. Sharing one would be a write-write race whenever + // two steps fail together. + var executionErr, syncAggregateErr error var executionRequestsRoot common.Hash // [New in Gloas:EIP7732] saved for envelope construction. // Always initialize for GLOAS so EncodeSSZ never sees nil sub-fields. @@ -1303,10 +1305,12 @@ func (a *ApiHandler) produceBeaconBody( defer func() { log.Info("BlockProduction: GetSyncAggregate took", "duration", time.Since(start)) }() - beaconBody.SyncAggregate, err = a.syncMessagePool.GetSyncAggregate(targetSlot-1, blockRoot) + aggregate, err := a.syncMessagePool.GetSyncAggregate(targetSlot-1, blockRoot) if err != nil { - executionErr = fmt.Errorf("produceBeaconBody: sync aggregate: %w", err) + syncAggregateErr = fmt.Errorf("produceBeaconBody: sync aggregate: %w", err) + return } + beaconBody.SyncAggregate = aggregate }) // Process operations all in parallel with each other. wg.Go(func() { @@ -1346,6 +1350,9 @@ func (a *ApiHandler) produceBeaconBody( if executionErr != nil { return nil, 0, executionErr } + if syncAggregateErr != nil { + return nil, 0, syncAggregateErr + } if executionPayload == nil { return nil, 0, errors.New("failed to produce execution payload") } diff --git a/cl/beacon/handler/block_production_test.go b/cl/beacon/handler/block_production_test.go index 0f0d08d47d4..fe1ebe99ea6 100644 --- a/cl/beacon/handler/block_production_test.go +++ b/cl/beacon/handler/block_production_test.go @@ -41,6 +41,7 @@ import ( "github.com/erigontech/erigon/cl/phase1/core/state" "github.com/erigontech/erigon/cl/phase1/core/state/lru" "github.com/erigontech/erigon/cl/phase1/execution_client" + sync_pool_mock "github.com/erigontech/erigon/cl/validator/sync_contribution_pool/mock_services" "github.com/erigontech/erigon/cl/validator/validator_params" "github.com/erigontech/erigon/common" "github.com/erigontech/erigon/common/hexutil" @@ -1058,6 +1059,29 @@ func TestProductionReportsAFailedCollectionExactlyOnce(t *testing.T) { require.Contains(t, captured, "boom") } +func TestProductionCollectsTwoFailingBodyStepsWithoutRacing(t *testing.T) { + ctrl := gomock.NewController(t) + _, _, _, _, postState, handler, _, _, _, _ := setupTestingHandler(t, clparams.ElectraVersion, log.Root(), false) + + engine := execution_client.NewMockExecutionEngine(ctrl) + engine.EXPECT().ForkChoiceUpdate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()). + Return([]byte{1, 2, 3, 4, 5, 6, 7, 8}, nil).AnyTimes() + engine.EXPECT().GetAssembledBlock(gomock.Any(), gomock.Any(), gomock.Any()). + Return(nil, nil, nil, nil, errors.New("boom")).AnyTimes() + engine.EXPECT().SupportInsertion().Return(true).AnyTimes() + handler.engine = engine + + // The body steps run concurrently, so each needs somewhere of its own to put its failure. + syncPool := sync_pool_mock.NewMockSyncContributionPool(ctrl) + syncPool.EXPECT().GetSyncAggregate(gomock.Any(), gomock.Any()). + Return(nil, errors.New("no aggregate")).AnyTimes() + handler.syncMessagePool = syncPool + + _, err := handler.produceBlock(t.Context(), 1, postState.Slot(), common.Hash{0x41}, postState, + postState.Slot()+1, common.Bytes96{}, common.Hash{}) + require.Error(t, err) +} + func TestProductionSaysNothingWhenTheRequestWasAbandoned(t *testing.T) { logs := captureProductionLogs(t) ctx, cancel := context.WithCancel(t.Context())