diff --git a/cl/beacon/handler/block_production.go b/cl/beacon/handler/block_production.go index 3d156ef8dbc..4f4ba1405c8 100644 --- a/cl/beacon/handler/block_production.go +++ b/cl/beacon/handler/block_production.go @@ -289,6 +289,40 @@ 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{} +} + +// 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) } @@ -297,16 +331,17 @@ 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), -) (*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: } } @@ -314,29 +349,70 @@ func pollAssembledPayload( defer deadlineTimer.Stop() retryTicker := time.NewTicker(retryTime) defer retryTicker.Stop() + + var ( + attempts int + failures int + firstErr error + ) 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, 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 { - 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 { - 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 { @@ -559,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 } @@ -677,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 ( @@ -743,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 @@ -755,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, @@ -1020,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. @@ -1037,10 +1116,10 @@ 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) + executionErr = fmt.Errorf("produceBeaconBody: expected withdrawals: %w", err) return } slotNumber := hexutil.Uint64(targetSlot) @@ -1064,7 +1143,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 { @@ -1073,10 +1152,11 @@ 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, 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 @@ -1089,28 +1169,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 } @@ -1177,7 +1257,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) } @@ -1225,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 { - log.Error("BlockProduction: Failed to get sync aggregate", "err", err) + syncAggregateErr = fmt.Errorf("produceBeaconBody: sync aggregate: %w", err) + return } + beaconBody.SyncAggregate = aggregate }) // Process operations all in parallel with each other. wg.Go(func() { @@ -1268,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 fddce8a9234..fe1ebe99ea6 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,10 @@ 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" + 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" "github.com/erigontech/erigon/common/log/v3" @@ -383,26 +388,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, _, _, _, ok := pollAssembledPayload(context.Background(), 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.True(t, ok) + 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, _, _, _, ok := pollAssembledPayload(context.Background(), 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 { @@ -410,17 +417,18 @@ func TestPollAssembledPayloadRetriesWhileBusy(t *testing.T) { } return want, nil, nil, nil, nil }) - require.True(t, ok) + 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, _, _, _, ok := pollAssembledPayload(context.Background(), 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 { @@ -428,35 +436,37 @@ func TestPollAssembledPayloadRetriesOnError(t *testing.T) { } return want, nil, nil, nil, nil }) - require.True(t, ok) + 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, _, _, _, ok := pollAssembledPayload(context.Background(), 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.False(t, ok) + 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 - _, _, _, _, ok := pollAssembledPayload(context.Background(), 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.False(t, ok) + require.Error(t, err) require.Equal(t, 1, calls) } @@ -466,12 +476,12 @@ func TestPollAssembledPayloadReturnsOnContextCancel(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() calls := 0 - _, _, _, _, ok := pollAssembledPayload(ctx, 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.False(t, ok) + require.Error(t, err) require.Zero(t, calls) } @@ -829,3 +839,259 @@ 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 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{} + previous := log.Root().GetHandler() + log.Root().SetHandler(log.StreamHandler(output, log.LogfmtFormat())) + t.Cleanup(func() { log.Root().SetHandler(previous) }) + return func() string { + var loud []string + for line := range strings.SplitSeq(output.String(), "\n") { + if strings.Contains(line, "lvl=eror") || strings.Contains(line, "lvl=warn") { + loud = append(loud, line) + } + } + return strings.Join(loud, "\n") + } +} + +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, _, _, _, err := 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("execution module is busy") + } + return &cltypes.Eth1Block{}, &engine_types.BlobsBundle{}, nil, big.NewInt(1), nil + }) + + 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 + _, _, _, _, 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 + }) + + require.NotZero(t, calls) + + // 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, err, boom) + require.Contains(t, err.Error(), "attempt") + require.Empty(t, logs(), "the poll does not report; its caller does") +} + +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)} + + _, _, _, _, 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 + }) + + // 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, err) + require.NotContains(t, logs(), "lvl=eror") +} + +func TestPollAssembledPayloadStillReportsFailuresThatPrecededTheCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + window := blockBuilderWindow{firstGetAt: time.Now(), pollUntil: time.Now().Add(time.Minute)} + + calls := 0 + _, _, _, _, err := 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. Reporting only the + // cancellation would lose the only sign of it. + require.NotErrorIs(t, err, context.Canceled) + require.Contains(t, err.Error(), "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")) +} + +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 + _, _, _, _, 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 + // 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.Error(t, err) + 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) { + ctx := t.Context() + logs := captureProductionLogs(t) + + 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 + // 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 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()) + 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()) +} 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)) }},